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
from flask import has_request_context, request
# 浏览器 WebSocket 必须走对外 host + /cw，不能注入容器内 127.0.0.1:3000
if has_request_context() and getattr(request, "host", None):
scheme = "wss" if request.scheme == "https" else "ws"
return f"{scheme}://{request.host}/cw"
public_base = (
CONFIG.chatwoot_guest_base_url
or CONFIG.chatwoot_base_url
or "http://127.0.0.1:7788/cw"
)
parsed = urlparse(public_base)
scheme = "wss" if parsed.scheme == "https" else "ws"
host = parsed.netloc or "127.0.0.1:7788"
path = (parsed.path or "/cw").rstrip("/") or "/cw"
return f"{scheme}://{host}{path}"
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")
# 避免瀏覽器快取舊版 widget HTML（導致注入腳本不更新）
response_headers = [
(key, value)
for key, value in response_headers
if key.lower()
not in {
"content-length",
"content-encoding",
"etag",
"last-modified",
"cache-control",
"pragma",
"expires",
}
]
response_headers.extend(
[
("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0"),
("Pragma", "no-cache"),
("Expires", "0"),
]
)
return Response(payload, status=status, headers=response_headers)