#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI 头脑风暴 — Flask 网站
=========================
- 主页: AI 头脑风暴（多风格）
- 工具区: 地址校验工具（元宝AI + DeepSeek 双引擎）
- 并发: Waitress + 10 线程
启动:
python app.py # 开发 / 生产 (自动选)
python app.py --dev # 强制开发模式
访问:
http://localhost:8765
"""
import io, json, os, re, sys, time, urllib.parse, tempfile, hashlib, shutil
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
from concurrent.futures import ThreadPoolExecutor, as_completed
import cloudscraper, pandas as pd, requests
from flask import Flask, request, jsonify, render_template, render_template_string, send_file, send_from_directory
from werkzeug.utils import secure_filename
from chat_routes import chat_bp, inject_chat_config
from video_routes import video_bp
# ============================================================
# 路径 & 配置
# ============================================================
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
TEMPLATE_DIR = os.path.join(SCRIPT_DIR, "templates")
CONFIG_FILE = os.path.join(SCRIPT_DIR, "engine_config.json")
KEYS_FILE = os.path.join(SCRIPT_DIR, "keys.txt")
PORT = 8765
app = Flask(__name__, template_folder=TEMPLATE_DIR, static_folder=SCRIPT_DIR, static_url_path="")
app.config['TEMPLATES_AUTO_RELOAD'] = True
app.jinja_env.auto_reload = True
@app.after_request
def disable_cache(response):
response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
response.headers["Pragma"] = "no-cache"
response.headers["Expires"] = "0"
# 打印请求日志（时间 方法 路径 → 状态码）
ts = time.strftime("%H:%M:%S")
print(f"[{ts}] {request.method} {request.path} → {response.status_code}", flush=True)
return response
# ============================================================
# 加载引擎配置
# ============================================================
def load_config():
default = {
"primary_engine": "deepseek",
"models": {"yuanbao": "hy3-preview", "deepseek": "deepseek-chat"}
}
if not os.path.exists(CONFIG_FILE):
return default
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
cfg = json.load(f)
for k in default:
cfg.setdefault(k, default[k])
return cfg
except Exception:
return default
config = load_config()
YUANBAO_MODEL = config.get("models", {}).get("yuanbao", "hy3-preview")
DEEPSEEK_MODEL = config.get("models", {}).get("deepseek", "deepseek-chat")
DOUBAO_MODEL = config.get("models", {}).get("doubao", "doubao-seed-1-6-flash-250828")
PPT_DOWNLOAD_DIR = config.get("ppt_download_dir", None) or os.path.join(os.path.expanduser("~"), "Downloads")
# 文件名 → 实际保存目录的映射（支持前端自定义路径）
_PPT_SAVE_DIR_MAP = {}
PRIMARY_ENGINE = config.get("primary_engine", "yuanbao")
# ============================================================
# 读取 API Keys
# ============================================================
def read_keys():
try:
with open(KEYS_FILE, "r", encoding="utf-8") as f:
content = f.read()
except FileNotFoundError:
return None, None, None, f"密钥文件不存在: {KEYS_FILE}"
except Exception as e:
return None, None, None, f"读取密钥文件失败: {e}"
yuanbao_key = None
deepseek_key = None
doubao_key = None
lines = content.splitlines()
in_deepseek_section = False
in_yuanbao_section = False
in_huoshansection = False
for line in lines:
line = line.strip()
if "元宝" in line or "yuanbao" in line.lower():
in_yuanbao_section = True
in_deepseek_section = False
in_huoshansection = False
continue
if line.lower() == "deepseek":
in_deepseek_section = True
in_yuanbao_section = False
in_huoshansection = False
continue
if "火山方舟" in line:
in_huoshansection = True
in_yuanbao_section = False
in_deepseek_section = False
continue
if in_yuanbao_section and line.startswith("sk-") and not yuanbao_key:
yuanbao_key = line
continue
if in_deepseek_section and line.startswith("sk-") and not deepseek_key:
deepseek_key = line
continue
if in_huoshansection and line.startswith("ark-") and not doubao_key:
doubao_key = line
continue
if not deepseek_key:
for line in lines:
line = line.strip()
if line.startswith("sk-") and line != yuanbao_key and not deepseek_key:
deepseek_key = line
break
# 如果没有在火山方舟段找到，全文件搜索 ark- key
if not doubao_key:
for line in lines:
line = line.strip()
if line.startswith("ark-") and not doubao_key:
doubao_key = line
break
errors = []
if not yuanbao_key:
errors.append("未找到元宝API Key")
if not deepseek_key:
errors.append("未找到DeepSeek API Key")
if not doubao_key:
errors.append("未找到火山方舟(豆包)API Key")
return yuanbao_key, deepseek_key, doubao_key, "; ".join(errors) if errors else None
YUANBAO_API_KEY, DEEPSEEK_API_KEY, DOUBAO_API_KEY, keys_error = read_keys()
YUANBAO_API_URL = "https://tokenhub.tencentmaas.com/v1/chat/completions"
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
DOUBAO_API_URL = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
# 注册聊天机器人蓝图
app.register_blueprint(chat_bp)
inject_chat_config({
"YUANBAO_API_URL": YUANBAO_API_URL,
"YUANBAO_API_KEY": YUANBAO_API_KEY,
"YUANBAO_MODEL": YUANBAO_MODEL,
"DEEPSEEK_API_URL": DEEPSEEK_API_URL,
"DEEPSEEK_API_KEY": DEEPSEEK_API_KEY,
"DEEPSEEK_MODEL": DEEPSEEK_MODEL,
})
# 注册视频生成蓝图
app.register_blueprint(video_bp)
# ============================================================
# 引擎调用 — 元宝AI
# ============================================================
HUNYUAN_PROMPT = """你是一位中国地理地址校验专家。请分析以下中文地址，判断其各个组成部分是否真实存在，并给出整体结论。
地址：{address}
请严格按以下格式输出，不要添加额外说明：
===地址分解===
（列出地址的每个层级，每行一条，格式：层级名称：具体内容（真实）或 层级名称：具体内容（虚构）或 层级名称：具体内容（不确定））
===整体结论===
真实存在 / 部分真实存在 / 不存在
===分析说明===
（简要说明判断依据，150字以内）
判断规则：
- "真实存在"：地址的所有层级都真实存在
- "部分真实存在"：地址部分层级真实存在，部分虚构或不存在
- "不存在"：地址核心层级本身虚构或完全无法对应真实地理信息
"""
def call_hunyuan(address):
payload = json.dumps({
"model": YUANBAO_MODEL,
"messages": [{"role": "user", "content": HUNYUAN_PROMPT.format(address=address)}],
"stream": False, "max_tokens": 2048, "temperature": 0.1
}, ensure_ascii=False).encode("utf-8")
req = Request(YUANBAO_API_URL, data=payload, headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {YUANBAO_API_KEY}"
}, method="POST")
try:
with urlopen(req, timeout=60) as resp:
data = json.loads(resp.read().decode("utf-8"))
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
if not content:
return {"success": False, "error": "API返回内容为空"}
return {"success": True, "raw": content}
except HTTPError as e:
err_body = e.read().decode("utf-8", errors="replace")
return {"success": False, "error": f"HTTP {e.code}: {err_body[:400]}"}
except URLError as e:
return {"success": False, "error": f"网络连接失败: {e.reason}"}
except Exception as e:
return {"success": False, "error": f"未知错误: {e}"}
def parse_hunyuan(raw_text):
parts = []
m_parts = re.search(r'===地址分解===\s*(.*?)(?:\s*===整体结论===|$)', raw_text, re.DOTALL)
if m_parts:
for line in m_parts.group(1).strip().split('\n'):
line = line.strip()
if not line:
continue
m = re.match(r'(.+?)[：:]\s*(.+?)[（(](真实|虚构|不确定)[）)]\s*$', line)
if m:
parts.append({"level": m.group(1).strip(), "name": m.group(2).strip(), "status": m.group(3)})
continue
m2 = re.match(r'(.+?)[（(](真实|虚构|不确定)[）)]\s*$', line)
if m2:
parts.append({"level": "", "name": m2.group(1).strip(), "status": m2.group(2)})
overall = "未知"
m_overall = re.search(r'===整体结论===\s*(.*?)(?:\s*===分析说明===|$)', raw_text, re.DOTALL)
if m_overall:
c = m_overall.group(1).strip()
if "真实存在" in c and "部分" not in c and "不" not in c:
overall = "真实存在"
elif "部分真实" in c:
overall = "部分真实存在"
elif "不存在" in c:
overall = "不存在"
analysis = ""
m_analysis = re.search(r'===分析说明===\s*(.*)', raw_text, re.DOTALL)
if m_analysis:
analysis = m_analysis.group(1).strip()
return {"parts": parts, "overall": overall, "analysis": analysis}
# ============================================================
# 引擎调用 — DeepSeek
# ============================================================
SYSTEM_DEEPSEEK = """你是一个中文地址校验专家，服务于金融反洗钱(AML)合规场景。
你的任务：
1. 判断给定地址是否真实、完整、规范
2. 将地址标准化（补齐省市区、统一格式）
3. 返回 JSON 格式结果
输出必须是纯 JSON，不要有任何额外文字:
{
"address_raw": "原始输入",
"address_standard": "标准化后的完整地址",
"is_valid": true/false,
"completeness": "完整/部分缺失/严重缺失",
"confidence": 0.0-1.0,
"issues": ["问题1", "问题2"],
"parsed": { "province": "省", "city": "市", "district": "区/县", "street": "街道/路", "detail": "详细地址" },
"risk_flags": ["风险标记"],
"suggestion": "修正建议或备注"
}
风险标记规则:
- 地址不存在或明显虚构 → "疑似虚构地址"
- 省市区不匹配 → "行政区划不匹配"
- 缺少关键信息（无门牌号等）→ "地址信息不完整"
- 地址指向敏感区域 → "敏感区域"
"""
def call_deepseek(address):
payload = json.dumps({
"model": DEEPSEEK_MODEL,
"messages": [
{"role": "system", "content": SYSTEM_DEEPSEEK},
{"role": "user", "content": address}
],
"stream": False, "max_tokens": 2048, "temperature": 0.1
}, ensure_ascii=False).encode("utf-8")
req = Request(DEEPSEEK_API_URL, data=payload, headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {DEEPSEEK_API_KEY}"
}, method="POST")
try:
with urlopen(req, timeout=60) as resp:
data = json.loads(resp.read().decode("utf-8"))
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
if not content:
return {"success": False, "error": "API返回内容为空"}
return {"success": True, "raw": content}
except HTTPError as e:
err_body = e.read().decode("utf-8", errors="replace")
return {"success": False, "error": f"HTTP {e.code}: {err_body[:400]}"}
except URLError as e:
return {"success": False, "error": f"网络连接失败: {e.reason}"}
except Exception as e:
return {"success": False, "error": f"未知错误: {e}"}
def parse_deepseek(raw_text):
text = raw_text.strip()
if text.startswith("```"):
text = re.sub(r"^```(?:json)?\s*", "", text)
text = re.sub(r"\s*```$", "", text)
try:
result = json.loads(text)
except json.JSONDecodeError:
return {"parts": [], "overall": "解析失败", "analysis": f"DeepSeek返回内容未能解析为JSON", "detail": {}}
is_valid = result.get("is_valid", False)
completeness = result.get("completeness", "")
overall = "真实存在" if is_valid else ("部分真实存在" if completeness == "部分缺失" else "不存在")
parts = []
parsed = result.get("parsed", {})
for key, label in [("province","省/直辖市"),("city","城市"),("district","区/县"),("street","街道/路")]:
if parsed.get(key):
parts.append({"level": label, "name": parsed[key], "status": "真实"})
if parsed.get("detail"):
parts.append({"level": "详细地址", "name": parsed["detail"], "status": "真实"})
analysis_parts = []
issues = result.get("issues", [])
if issues:
analysis_parts.append("问题: " + "; ".join(issues))
flags = result.get("risk_flags", [])
if flags:
analysis_parts.append("风险: " + "; ".join(f for f in flags if f))
if result.get("suggestion"):
analysis_parts.append(f"建议: {result['suggestion']}")
if result.get("confidence"):
analysis_parts.append(f"置信度: {result['confidence']:.0%}")
analysis = " | ".join(analysis_parts) if analysis_parts else "DeepSeek分析完成"
return {
"parts": parts,
"overall": overall,
"analysis": analysis,
"detail": {
"address_standard": result.get("address_standard", ""),
"completeness": result.get("completeness", ""),
"confidence": result.get("confidence", 0),
"risk_flags": result.get("risk_flags", []),
"issues": result.get("issues", []),
"suggestion": result.get("suggestion", ""),
}
}
# ============================================================
# 引擎调用 — 翻译（双引擎通用）
# ============================================================
def call_engine_for_translate(engine, prompt, system_prompt=None):
if engine == "deepseek":
url = DEEPSEEK_API_URL
key = DEEPSEEK_API_KEY
messages = [{"role": "user", "content": prompt}]
if system_prompt:
messages.insert(0, {"role": "system", "content": system_prompt})
payload = json.dumps({
"model": DEEPSEEK_MODEL,
"messages": messages,
"stream": False, "max_tokens": 4096, "temperature": 0.1
}, ensure_ascii=False).encode("utf-8")
else:
url = YUANBAO_API_URL
key = YUANBAO_API_KEY
payload = json.dumps({
"model": YUANBAO_MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": False, "max_tokens": 4096, "temperature": 0.1
}, ensure_ascii=False).encode("utf-8")
req = Request(url, data=payload, headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {key}"
}, method="POST")
try:
with urlopen(req, timeout=60) as resp:
data = json.loads(resp.read().decode("utf-8"))
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
if not content:
return {"success": False, "error": "API返回内容为空"}
return {"success": True, "raw": content}
except HTTPError as e:
err_body = e.read().decode("utf-8", errors="replace")
return {"success": False, "error": f"HTTP {e.code}: {err_body[:400]}"}
except URLError as e:
return {"success": False, "error": f"网络连接失败: {e.reason}"}
except Exception as e:
return {"success": False, "error": f"未知错误: {e}"}
LANGUAGE_MAP = {
"英语": "English", "法语": "French", "德语": "German",
"日语": "Japanese", "俄语": "Russian", "西班牙语": "Spanish",
"葡萄牙语": "Portuguese", "中文": "Chinese",
}
def _extract_json(raw):
"""从 LLM 响应中尽可能提取 JSON 对象，支持各种异常格式。"""
s = raw.strip()
# 清理代码块标记
s = re.sub(r"^```(?:json)?\s*", "", s)
s = re.sub(r"\s*```$", "", s)
s = s.strip()
# 尝试直接解析
try:
return json.loads(s)
except json.JSONDecodeError:
pass
# 尝试提取最外层大括号内容
m = re.search(r'\{.*\}', s, re.DOTALL)
if m:
candidate = m.group(0)
try:
return json.loads(candidate)
except json.JSONDecodeError:
pass
# 尝试修复常见问题：多余的逗号
candidate = re.sub(r',\s*}', '}', candidate)
candidate = re.sub(r',\s*]', ']', candidate)
try:
return json.loads(candidate)
except json.JSONDecodeError:
pass
# 尝试修复引号被转义的问题
candidate = candidate.replace('\\"', '"')
candidate = candidate.replace("\\'", "'")
try:
return json.loads(candidate)
except json.JSONDecodeError:
pass
# 尝试按行解析：语言: 文本 格式
lines = [l.strip() for l in s.split("\n") if l.strip()]
result = {"translations": []}
for line in lines:
for lang_name in LANGUAGE_MAP:
if lang_name in line:
# 提取语言名后的内容
idx = line.index(lang_name) + len(lang_name)
text_part = line[idx:].lstrip("：:）（)")
if text_part:
result["translations"].append({"language": lang_name, "text": text_part})
break
if result["translations"]:
return result
return None
@app.route("/api/translate", methods=["POST"])
def api_translate():
body = request.get_json(silent=True) or {}
text = (body.get("text") or "").strip()
if not text:
return jsonify({"error": "请输入要翻译的文本"}), 400
if len(text) > 2000:
return jsonify({"error": "文本过长（最多2000字符）"}), 400
engine = body.get("engine", "deepseek")
if engine not in ("yuanbao", "deepseek"):
engine = "yuanbao"
target_languages = body.get("target_languages", [])
if not isinstance(target_languages, list):
target_languages = [target_languages] if target_languages else []
# 自动检测源语言
has_chinese = bool(re.search(r'[\u4e00-\u9fff]', text))
# 默认目标语言
if not target_languages:
if has_chinese:
target_languages = ["英语"]
else:
target_languages = ["中文"]
# 去重
seen = set()
unique_langs = []
for lang in target_languages:
if lang not in seen:
seen.add(lang)
unique_langs.append(lang)
target_languages = unique_langs
# 校验 API Key
if engine == "deepseek" and not DEEPSEEK_API_KEY:
return jsonify({"error": "DeepSeek API Key 未配置", "engine": engine}), 400
if engine == "yuanbao" and not YUANBAO_API_KEY:
return jsonify({"error": "元宝API Key 未配置", "engine": engine}), 400
# 构建翻译 prompt — 严格 JSON 格式
lang_str = "、".join(target_languages)
example_pairs = [f'{{"language": "{lang}", "text": ""}}' for lang in target_languages]
example_json = "[" + ", ".join(example_pairs) + "]"
lang_hints = " / ".join([f"{i+1}.{lang}" for i, lang in enumerate(target_languages)])
prompt = f"""请将以下文本翻译成指定的目标语言。
原文：{text}
目标语言：{lang_str}
请严格按照 JSON 格式输出，不要加任何额外文字、不要加代码块标记：
{{"translations": {example_json}}}"""
if engine == "deepseek":
prompt = f"你是一个翻译助手，只输出JSON。\n\n{prompt}"
t0 = time.time()
result = call_engine_for_translate(engine, prompt)
elapsed = round(time.time() - t0, 2)
if not result["success"]:
return jsonify({"error": result["error"], "engine": engine}), 500
raw = result["raw"].strip()
# 尝试解析 JSON
parsed = _extract_json(raw)
if parsed is None:
# 最后兜底：直接按语言顺序把每行翻译当作结果
lines = [l for l in raw.split("\n") if l.strip()]
translations = []
for i, lang in enumerate(target_languages):
if i < len(lines):
translations.append({"language": lang, "text": lines[i].strip()})
if translations:
return jsonify({
"success": True,
"engine": engine,
"source_text": text,
"is_chinese_source": has_chinese,
"translations": translations,
"elapsed": elapsed,
})
return jsonify({"error": "翻译结果解析失败", "raw": raw[:500], "engine": engine, "elapsed": elapsed}), 500
translations = parsed.get("translations", [])
if not translations:
# 有时 LLM 直接以语言名为 key 平铺
for lang in target_languages:
if lang in parsed:
translations.append({"language": lang, "text": parsed[lang]})
if not translations:
return jsonify({"error": "翻译结果格式异常", "raw": raw[:500], "engine": engine, "elapsed": elapsed}), 500
return jsonify({
"success": True,
"engine": engine,
"source_text": text,
"is_chinese_source": has_chinese,
"translations": translations,
"elapsed": elapsed,
})
# ============================================================
# Flask 路由 — 页面
# ============================================================
@app.route("/")
def index():
return render_template("index.html",
yuanbao_ready=bool(YUANBAO_API_KEY),
deepseek_ready=bool(DEEPSEEK_API_KEY),
primary_engine=PRIMARY_ENGINE,
yuanbao_model=YUANBAO_MODEL,
deepseek_model=DEEPSEEK_MODEL,
keys_status="正常" if (YUANBAO_API_KEY or DEEPSEEK_API_KEY) else "未配置",
)
@app.route("/game/pacman")
def game_pacman():
return render_template("pacman.html")
@app.route("/readme")
def readme():
readme_path = os.path.join(SCRIPT_DIR, "README.md")
if not os.path.exists(readme_path):
return "README.md 文件不存在", 404
with open(readme_path, "r", encoding="utf-8") as f:
content = f.read()
html_body = []
in_table = False; in_code = False; buf = []; rows = []
for line in content.split("\n"):
if line.startswith("```"):
if in_code:
html_body.append(f'
{"".join(buf)}
'); buf = []; in_code = False
else: in_code = True
continue
if in_code: buf.append(line + "\n"); continue
if line.startswith("|") and line.endswith("|"):
cells = [c.strip() for c in line.split("|")[1:-1]]
if not in_table: in_table = True; rows = [cells]
elif all(c.startswith("-") for c in "".join(cells) if c): continue
else: rows.append(cells)
continue
if in_table and rows:
html_body.append("
" + "".join(f"
{c}
" for c in rows[0]) + "
")
for r in rows[1:]: html_body.append("
" + "".join(f"
{c}
" for c in r) + "
")
html_body.append("
"); rows = []; in_table = False
if not line.strip(): html_body.append(" "); continue
if line.strip() == "---": html_body.append(""); continue
t = line
t = t.replace("&","&").replace("<","<").replace(">",">")
import re as _re
t = _re.sub(r'\*\*(.+?)\*\*', r'\1', t)
t = _re.sub(r'`([^`]+)`', r'\1', t)
t = _re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'\1', t)
if line.startswith("### "): html_body.append(f"