from __future__ import annotations
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from flask import Response, request
from config import CONFIG
HOP_BY_HOP = {
"connection",
"keep-alive",
"proxy-authenticate",
"proxy-authorization",
"te",
"trailers",
"transfer-encoding",
"upgrade",
"host",
"content-length",
}
INJECT_SCRIPT = """
"""
def _cable_ws_origin() -> str:
from urllib.parse import urlparse
parsed = urlparse(CONFIG.chatwoot_guest_proxy_target or "http://127.0.0.1:3000")
scheme = "wss" if parsed.scheme == "https" else "ws"
host = parsed.netloc or "127.0.0.1:3000"
return f"{scheme}://{host}"
def _inject_script() -> str:
return INJECT_SCRIPT.replace("__CABLE_WS_ORIGIN__", _cable_ws_origin())
def _forward_headers() -> dict[str, str]:
headers: dict[str, str] = {}
for key, value in request.headers.items():
if key.lower() in HOP_BY_HOP:
continue
headers[key] = value
return headers
def proxy_chatwoot(subpath: str) -> Response:
target = CONFIG.chatwoot_guest_proxy_target.rstrip("/")
query = urlencode(request.args, doseq=True)
url = f"{target}/{subpath.lstrip('/')}"
if query:
url = f"{url}?{query}"
body = request.get_data() if request.method not in {"GET", "HEAD"} else None
forwarded = Request(
url,
data=body,
headers=_forward_headers(),
method=request.method,
)
try:
with urlopen(forwarded, timeout=30) as upstream:
payload = upstream.read()
status = upstream.status
content_type = upstream.headers.get("Content-Type", "")
response_headers = [
(key, value)
for key, value in upstream.headers.items()
if key.lower() not in HOP_BY_HOP
]
except HTTPError as exc:
payload = exc.read()
status = exc.code
content_type = exc.headers.get("Content-Type", "") if exc.headers else ""
response_headers = []
except URLError as exc:
return Response(
f"无法连线本机 Chatwoot：{exc.reason}",
status=502,
mimetype="text/plain; charset=utf-8",
)
if "text/html" in content_type.lower() and payload:
html = payload.decode("utf-8", "replace")
inject = _inject_script()
if "" in html:
html = html.replace("", inject + "", 1)
elif "" in html:
html = html.replace("", inject + "", 1)
else:
html += inject
payload = html.encode("utf-8")
return Response(payload, status=status, headers=response_headers)