#!/usr/bin/env python3 """Local Hongcan article library and Qwen-assisted editorial analysis.""" from __future__ import annotations import json import os import sqlite3 import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from dotenv import load_dotenv from flask import Flask, abort, jsonify, render_template, request BASE_DIR = Path(__file__).resolve().parent DB_PATH = Path(os.getenv("HONGCAN_DB", BASE_DIR / "data" / "hongcan.db")) load_dotenv(BASE_DIR / ".env") app = Flask(__name__) app.secret_key = os.getenv("FLASK_SECRET_KEY", "hongcan-local-only") def now_iso() -> str: return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") def db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") return conn def ensure_schema() -> None: with db() as conn: conn.executescript((BASE_DIR / "schema.sql").read_text(encoding="utf-8")) def clamp_int(value: str | None, default: int, low: int, high: int) -> int: try: return max(low, min(high, int(value or default))) except ValueError: return default TREND_CN = {"emerging": "新兴", "accelerating": "加速", "stable": "稳定", "declining": "下行"} @app.template_filter("date_cn") def date_cn(value: str | None) -> str: if not value: return "时间未知" try: return datetime.fromisoformat(value).strftime("%Y.%m.%d %H:%M") except ValueError: return value @app.get("/") def index(): query = (request.args.get("q") or "").strip() page = clamp_int(request.args.get("page"), 1, 1, 10_000) per_page = 18 where = "WHERE crawl_status='success'" params: list[object] = [] if query: where += " AND (title LIKE ? OR author LIKE ? OR content LIKE ?)" needle = f"%{query}%" params.extend([needle, needle, needle]) with db() as conn: total = conn.execute(f"SELECT COUNT(*) FROM articles {where}", params).fetchone()[0] rows = conn.execute( f""" SELECT a.id,a.title,a.author,a.published_at,a.summary,a.canonical_url, substr(a.content,1,180) AS excerpt, json_extract(x.result_json,'$.summary') AS ai_summary, CASE WHEN x.id IS NULL THEN 0 ELSE 1 END AS analyzed FROM articles a LEFT JOIN ai_analyses x ON x.article_id=a.id AND x.analysis_type='editorial' {where.replace('crawl_status', 'a.crawl_status')} ORDER BY a.published_at DESC,a.id DESC LIMIT ? OFFSET ? """, [*params, per_page, (page - 1) * per_page], ).fetchall() stats = conn.execute( """ SELECT COUNT(*) total, SUM(crawl_status='success') success, MAX(published_at) latest, (SELECT COUNT(DISTINCT article_id) FROM ai_analyses) analyzed FROM articles """ ).fetchone() signal_rows = conn.execute( "SELECT * FROM industry_signals ORDER BY signal_date DESC,confidence DESC LIMIT 8" ).fetchall() signals = [] for signal in signal_rows: item = dict(signal) item["tags"] = json.loads(item["tags_json"]) item["implications"] = json.loads(item["implications_json"]) item["article_ids"] = json.loads(item["article_ids_json"]) item["trend"] = TREND_CN.get(item["trend"], item["trend"]) signals.append(item) return render_template( "index.html", articles=rows, query=query, page=page, total=total, pages=max(1, (total + per_page - 1) // per_page), stats=stats, signals=signals, ) @app.get("/signals/") def signal_detail(signal_id: int): with db() as conn: signal = conn.execute("SELECT * FROM industry_signals WHERE id=?", (signal_id,)).fetchone() if not signal: abort(404) signal = dict(signal) signal["trend"] = TREND_CN.get(signal["trend"], signal["trend"]) article_ids = json.loads(signal["article_ids_json"]) articles = [] if article_ids: marks = ",".join("?" for _ in article_ids) articles = conn.execute( f"SELECT id,title,published_at FROM articles WHERE id IN ({marks}) ORDER BY published_at DESC", article_ids, ).fetchall() return render_template( "signal.html", signal=signal, articles=articles, implications=json.loads(signal["implications_json"]), tags=json.loads(signal["tags_json"]), ) @app.get("/article/") def article(article_id: int): with db() as conn: row = conn.execute("SELECT * FROM articles WHERE id=?", (article_id,)).fetchone() if not row: abort(404) analysis = conn.execute( "SELECT * FROM ai_analyses WHERE article_id=? AND analysis_type='editorial' ORDER BY updated_at DESC LIMIT 1", (article_id,), ).fetchone() parsed = json.loads(analysis["result_json"]) if analysis else None return render_template("article.html", article=row, analysis=parsed, analysis_row=analysis) def qwen_analyze(title: str, content: str) -> dict: api_key = os.getenv("DASHSCOPE_API_KEY", "").strip() if not api_key: raise RuntimeError("尚未配置 DASHSCOPE_API_KEY，请在 .env 中设置") model = os.getenv("QWEN_MODEL", "qwen-plus") prompt = f"""你是一名资深餐饮行业研究编辑。分析下面这篇文章，只返回合法 JSON，不要 Markdown 代码块。 JSON 字段必须为：summary（120字内摘要）、key_points（3-6条字符串）、industry_signal（行业信号）、brands（品牌数组）、numbers（关键数据数组）、content_angles（3条可延展选题）、risk_notes（事实或表述风险数组）、tags（5-10个标签）、sentiment（positive/neutral/negative）。 文章标题：{title} 正文：{content[:24000]}""" payload = json.dumps({ "model": model, "messages": [ {"role": "system", "content": "你输出严谨、简洁、可供编辑部直接使用的结构化行业分析。"}, {"role": "user", "content": prompt}, ], "temperature": 0.25, "response_format": {"type": "json_object"}, }, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions", data=payload, headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=90) as response: body = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace")[:500] raise RuntimeError(f"千问接口返回 {exc.code}：{detail}") from exc text = body["choices"][0]["message"]["content"].strip() if text.startswith("```"): text = text.strip("`").removeprefix("json").strip() return json.loads(text) @app.post("/api/articles//analyze") def analyze(article_id: int): force = bool((request.get_json(silent=True) or {}).get("force")) model = os.getenv("QWEN_MODEL", "qwen-plus") with db() as conn: row = conn.execute("SELECT * FROM articles WHERE id=?", (article_id,)).fetchone() if not row: return jsonify({"ok": False, "error": "文章不存在"}), 404 cached = conn.execute( "SELECT * FROM ai_analyses WHERE article_id=? AND model=? AND analysis_type='editorial'", (article_id, model), ).fetchone() if cached and cached["content_hash"] == row["content_hash"] and not force: return jsonify({"ok": True, "cached": True, "analysis": json.loads(cached["result_json"])}) try: result = qwen_analyze(row["title"], row["content"]) except Exception as exc: return jsonify({"ok": False, "error": str(exc)}), 502 timestamp = now_iso() conn.execute( """ INSERT INTO ai_analyses(article_id,model,analysis_type,result_json,content_hash,created_at,updated_at) VALUES (?,?, 'editorial', ?,?,?,?) ON CONFLICT(article_id,model,analysis_type) DO UPDATE SET result_json=excluded.result_json,content_hash=excluded.content_hash,updated_at=excluded.updated_at """, (article_id, model, json.dumps(result, ensure_ascii=False), row["content_hash"], timestamp, timestamp), ) conn.commit() return jsonify({"ok": True, "cached": False, "analysis": result}) if __name__ == "__main__": ensure_schema() app.run(host="127.0.0.1", port=int(os.getenv("PORT", "8766")), debug=False)