freedakgmail
868 lines
31 KiB
Python
868 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
"""餐智库后端 API 服务。
|
|
|
|
提供前端所需的全部接口：
|
|
- GET /api/dashboard 仪表盘数据
|
|
- GET /api/intel 情报流列表（支持筛选）
|
|
- GET /api/intel/<id> 情报详情
|
|
- GET /api/brands 品牌库列表
|
|
- GET /api/brands/<id> 品牌详情
|
|
- GET /api/analysis 赛道分析数据
|
|
- GET /api/reports 报告列表
|
|
- POST /api/qa/ask RAG 问答
|
|
- POST /api/crawl/run 触发采集
|
|
- POST /api/analyze/run 触发结构化分析
|
|
- GET /api/stats 概览统计
|
|
- GET /api/health 健康检查
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from flask import Flask, jsonify, request, send_file, abort, Response, stream_with_context
|
|
from flask_cors import CORS
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
load_dotenv(BASE_DIR / ".env")
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.getenv("FLASK_SECRET_KEY", "cibank-local-only")
|
|
CORS(app)
|
|
|
|
DB_PATH = Path(os.getenv("CIBANK_DB", BASE_DIR / "data" / "cibank.db"))
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
def db() -> sqlite3.Connection:
|
|
conn = sqlite3.connect(DB_PATH, timeout=10)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode = WAL")
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
conn.execute("PRAGMA busy_timeout = 5000")
|
|
return conn
|
|
|
|
|
|
def ensure_schema() -> None:
|
|
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
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
|
|
|
|
|
|
# ---- 健康检查 ----
|
|
|
|
@app.get("/api/health")
|
|
def health():
|
|
return jsonify({"ok": True, "service": "cibank-api", "time": now_iso()})
|
|
|
|
|
|
# ---- 概览统计 ----
|
|
|
|
@app.get("/api/stats")
|
|
def stats():
|
|
with db() as conn:
|
|
total_articles = conn.execute("SELECT COUNT(*) FROM articles WHERE crawl_status='success'").fetchone()[0]
|
|
total_analyzed = conn.execute("SELECT COUNT(DISTINCT article_id) FROM ai_analyses").fetchone()[0]
|
|
total_brands = conn.execute("SELECT COUNT(*) FROM brands").fetchone()[0]
|
|
total_reports = conn.execute("SELECT COUNT(*) FROM reports").fetchone()[0]
|
|
sources_mp = conn.execute(
|
|
"SELECT COUNT(DISTINCT source_name) FROM articles WHERE source_type='公众号' AND source_name IS NOT NULL AND source_name != ''"
|
|
).fetchone()[0]
|
|
sources_video = conn.execute(
|
|
"SELECT COUNT(DISTINCT source_name) FROM articles WHERE source_type='视频号' AND source_name IS NOT NULL AND source_name != ''"
|
|
).fetchone()[0]
|
|
today = datetime.now().astimezone().date().isoformat()
|
|
today_new = conn.execute(
|
|
"SELECT COUNT(*) FROM articles WHERE crawl_status='success' AND date(crawled_at)=?",
|
|
(today,),
|
|
).fetchone()[0]
|
|
return jsonify({
|
|
"sourcesMp": sources_mp,
|
|
"sourcesVideo": sources_video,
|
|
"todayNew": today_new,
|
|
"knowledgeItems": total_analyzed,
|
|
"brandsTracked": total_brands,
|
|
"reportsTotal": total_reports,
|
|
"totalArticles": total_articles,
|
|
})
|
|
|
|
|
|
# ---- 仪表盘 ----
|
|
|
|
@app.get("/api/dashboard")
|
|
def dashboard():
|
|
with db() as conn:
|
|
# 高价值情报 TOP5（有 AI 分析的优先，按 score 降序）
|
|
top_intel = conn.execute(
|
|
"""
|
|
SELECT a.id, a.title, a.source_name, a.source_type, a.published_at, a.category,
|
|
json_extract(x.result_json, '$.summary') AS ai_summary,
|
|
COALESCE(json_extract(x.result_json, '$.type'), '品牌动态') AS type,
|
|
COALESCE(CAST(json_extract(x.result_json, '$.score') AS INTEGER), 0) AS score,
|
|
json_extract(x.result_json, '$.brands') AS entities,
|
|
x.result_json IS NOT NULL AS has_analysis
|
|
FROM articles a
|
|
LEFT JOIN ai_analyses x ON x.article_id = a.id AND x.analysis_type = 'editorial'
|
|
WHERE a.crawl_status = 'success'
|
|
ORDER BY has_analysis DESC, score DESC, a.published_at DESC
|
|
LIMIT 5
|
|
"""
|
|
).fetchall()
|
|
|
|
# 政策预警
|
|
policy_alerts = conn.execute(
|
|
"""
|
|
SELECT a.id, a.title, a.source_name, a.published_at,
|
|
json_extract(x.result_json, '$.summary') AS ai_summary
|
|
FROM articles a
|
|
LEFT JOIN ai_analyses x ON x.article_id = a.id AND x.analysis_type = 'editorial'
|
|
WHERE a.crawl_status = 'success'
|
|
AND json_extract(x.result_json, '$.type') = '政策监管'
|
|
ORDER BY a.published_at DESC
|
|
LIMIT 3
|
|
"""
|
|
).fetchall()
|
|
|
|
# 热词统计（从标签中聚合）
|
|
tag_rows = conn.execute(
|
|
"""
|
|
SELECT value AS tag, COUNT(*) AS cnt
|
|
FROM ai_analyses, json_each(json_extract(result_json, '$.tags'))
|
|
GROUP BY value
|
|
ORDER BY cnt DESC
|
|
LIMIT 8
|
|
"""
|
|
).fetchall()
|
|
|
|
return jsonify({
|
|
"topIntel": [
|
|
{
|
|
"id": str(r["id"]),
|
|
"title": r["title"],
|
|
"source": r["source_name"] or r["source_type"],
|
|
"sourceType": r["source_type"],
|
|
"date": (r["published_at"] or "")[:10],
|
|
"type": r["type"] or "经营干货",
|
|
"score": r["score"] or 0,
|
|
"summary": r["ai_summary"] or "",
|
|
"entities": [b["name"] if isinstance(b, dict) else b for b in (json.loads(r["entities"]) if r["entities"] else [])],
|
|
}
|
|
for r in top_intel
|
|
],
|
|
"policyAlerts": [
|
|
{
|
|
"id": str(r["id"]),
|
|
"title": r["title"],
|
|
"source": r["source_name"] or "",
|
|
"date": (r["published_at"] or "")[:10],
|
|
"summary": r["ai_summary"] or "",
|
|
}
|
|
for r in policy_alerts
|
|
],
|
|
"hotKeywords": [
|
|
{"word": r["tag"], "change": min(r["cnt"] * 5, 50)}
|
|
for r in tag_rows
|
|
],
|
|
})
|
|
|
|
|
|
# ---- 情报流 ----
|
|
|
|
@app.get("/api/intel")
|
|
def intel_list():
|
|
type_filter = request.args.get("type", "")
|
|
category = request.args.get("category", "")
|
|
query = (request.args.get("q") or "").strip()
|
|
page = clamp_int(request.args.get("page"), 1, 1, 10_000)
|
|
per_page = clamp_int(request.args.get("per_page"), 20, 1, 100)
|
|
|
|
where = "WHERE a.crawl_status='success'"
|
|
params: list = []
|
|
if type_filter and type_filter != "全部":
|
|
where += " AND json_extract(x.result_json, '$.type') = ?"
|
|
params.append(type_filter)
|
|
if category and category != "全部赛道":
|
|
where += " AND a.category = ?"
|
|
params.append(category)
|
|
if query:
|
|
where += " AND (a.title LIKE ? OR a.summary LIKE ? OR a.content LIKE ?)"
|
|
needle = f"%{query}%"
|
|
params.extend([needle, needle, needle])
|
|
|
|
with db() as conn:
|
|
total = conn.execute(
|
|
f"""
|
|
SELECT COUNT(*) FROM articles a
|
|
LEFT JOIN ai_analyses x ON x.article_id = a.id AND x.analysis_type = 'editorial'
|
|
{where}
|
|
""",
|
|
params,
|
|
).fetchone()[0]
|
|
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT a.id, a.title, a.source_name, a.source_type, a.published_at, a.category,
|
|
a.summary, a.canonical_url,
|
|
json_extract(x.result_json, '$.summary') AS ai_summary,
|
|
json_extract(x.result_json, '$.key_points') AS key_points,
|
|
json_extract(x.result_json, '$.type') AS type,
|
|
json_extract(x.result_json, '$.score') AS score,
|
|
json_extract(x.result_json, '$.brands') AS entities,
|
|
json_extract(x.result_json, '$.timeliness') AS timeliness,
|
|
json_extract(x.result_json, '$.tags') AS tags
|
|
FROM articles a
|
|
LEFT JOIN ai_analyses x ON x.article_id = a.id AND x.analysis_type = 'editorial'
|
|
{where}
|
|
ORDER BY a.published_at DESC, a.id DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
[*params, per_page, (page - 1) * per_page],
|
|
).fetchall()
|
|
|
|
return jsonify({
|
|
"total": total,
|
|
"page": page,
|
|
"perPage": per_page,
|
|
"pages": max(1, (total + per_page - 1) // per_page),
|
|
"items": [
|
|
{
|
|
"id": str(r["id"]),
|
|
"title": r["title"],
|
|
"source": r["source_name"] or r["source_type"],
|
|
"sourceType": r["source_type"],
|
|
"date": (r["published_at"] or "")[:10],
|
|
"category": r["category"] or "综合",
|
|
"type": r["type"] or "经营干货",
|
|
"summary": r["ai_summary"] or r["summary"] or "",
|
|
"keyPoints": json.loads(r["key_points"]) if r["key_points"] else [],
|
|
"score": r["score"] or 0,
|
|
"entities": [b["name"] if isinstance(b, dict) else b for b in (json.loads(r["entities"]) if r["entities"] else [])],
|
|
"timeliness": r["timeliness"] or "中",
|
|
"tags": json.loads(r["tags"]) if r["tags"] else [],
|
|
"url": r["canonical_url"],
|
|
}
|
|
for r in rows
|
|
],
|
|
})
|
|
|
|
|
|
@app.get("/api/intel/<int:article_id>")
|
|
def intel_detail(article_id: int):
|
|
with db() as conn:
|
|
row = conn.execute("SELECT * FROM articles WHERE id=?", (article_id,)).fetchone()
|
|
if not row:
|
|
return jsonify({"ok": False, "error": "文章不存在"}), 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 jsonify({
|
|
"ok": True,
|
|
"article": {
|
|
"id": row["id"],
|
|
"title": row["title"],
|
|
"source": row["source_name"] or row["source_type"],
|
|
"sourceType": row["source_type"],
|
|
"author": row["author"],
|
|
"publishedAt": row["published_at"],
|
|
"summary": row["summary"],
|
|
"content": row["content"],
|
|
"url": row["canonical_url"],
|
|
"category": row["category"],
|
|
"mediaUrl": row["media_url"] if "media_url" in row.keys() else None,
|
|
},
|
|
"analysis": parsed,
|
|
})
|
|
|
|
|
|
# ---- 视频流 ----
|
|
|
|
@app.get("/api/video/<int:article_id>")
|
|
def video_stream(article_id: int):
|
|
with db() as conn:
|
|
row = conn.execute(
|
|
"SELECT media_url FROM articles WHERE id=? AND source_type='视频号'", (article_id,)
|
|
).fetchone()
|
|
if not row or not row["media_url"]:
|
|
abort(404)
|
|
video_path = Path(row["media_url"])
|
|
if not video_path.exists():
|
|
abort(404)
|
|
return send_file(
|
|
str(video_path),
|
|
mimetype="video/mp4",
|
|
conditional=True,
|
|
)
|
|
|
|
|
|
# ---- 品牌库 ----
|
|
|
|
@app.get("/api/brands")
|
|
def brand_list():
|
|
category = request.args.get("category", "")
|
|
query = (request.args.get("q") or "").strip()
|
|
page = max(1, request.args.get("page", 1, type=int))
|
|
per_page = min(100, max(1, request.args.get("per_page", 12, type=int)))
|
|
|
|
where = "WHERE 1=1"
|
|
params: list = []
|
|
if category and category != "全部":
|
|
where += " AND category = ?"
|
|
params.append(category)
|
|
if query:
|
|
where += " AND name LIKE ?"
|
|
params.append(f"%{query}%")
|
|
|
|
with db() as conn:
|
|
total = conn.execute(f"SELECT COUNT(*) FROM brands {where}", params).fetchone()[0]
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT * FROM brands {where}
|
|
ORDER BY stores DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
params + [per_page, (page - 1) * per_page],
|
|
).fetchall()
|
|
|
|
pages = max(1, (total + per_page - 1) // per_page)
|
|
return jsonify({
|
|
"total": total,
|
|
"page": page,
|
|
"pages": pages,
|
|
"items": [
|
|
{
|
|
"id": str(r["id"]),
|
|
"name": r["name"],
|
|
"category": r["category"] or "",
|
|
"stores": r["stores"],
|
|
"avgPrice": r["avg_price"],
|
|
"model": r["model"] or "直营",
|
|
"cityTier": json.loads(r["city_tier_json"]) if r["city_tier_json"] else [],
|
|
"trend": json.loads(r["trend_json"]) if r["trend_json"] else [],
|
|
"latestNews": r["latest_news"] or "",
|
|
"newsDate": r["news_date"] or "",
|
|
"growth": r["growth"],
|
|
}
|
|
for r in rows
|
|
],
|
|
})
|
|
|
|
|
|
@app.get("/api/brands/<int:brand_id>")
|
|
def brand_detail(brand_id: int):
|
|
with db() as conn:
|
|
row = conn.execute("SELECT * FROM brands WHERE id=?", (brand_id,)).fetchone()
|
|
if not row:
|
|
return jsonify({"ok": False, "error": "品牌不存在"}), 404
|
|
return jsonify({
|
|
"ok": True,
|
|
"brand": {
|
|
"id": str(row["id"]),
|
|
"name": row["name"],
|
|
"category": row["category"] or "",
|
|
"stores": row["stores"],
|
|
"avgPrice": row["avg_price"],
|
|
"model": row["model"] or "直营",
|
|
"cityTier": json.loads(row["city_tier_json"]) if row["city_tier_json"] else [],
|
|
"trend": json.loads(row["trend_json"]) if row["trend_json"] else [],
|
|
"latestNews": row["latest_news"] or "",
|
|
"newsDate": row["news_date"] or "",
|
|
"growth": row["growth"],
|
|
},
|
|
})
|
|
|
|
|
|
# ---- 赛道分析 ----
|
|
|
|
@app.get("/api/analysis")
|
|
def analysis():
|
|
with db() as conn:
|
|
# 赛道热度趋势：按月统计各赛道的文章数和平均评分，热度=文章数×平均评分/10
|
|
heat_rows = conn.execute(
|
|
"""
|
|
SELECT strftime('%Y-%m', a.published_at) AS month,
|
|
a.category,
|
|
COUNT(*) AS article_cnt,
|
|
AVG(CAST(json_extract(x.result_json, '$.score') AS REAL)) AS avg_score
|
|
FROM articles a
|
|
LEFT JOIN ai_analyses x ON x.article_id = a.id AND x.analysis_type = 'editorial'
|
|
WHERE a.crawl_status = 'success' AND a.category IS NOT NULL AND a.category != ''
|
|
AND a.published_at >= date('now', '-12 months')
|
|
GROUP BY month, a.category
|
|
ORDER BY month
|
|
"""
|
|
).fetchall()
|
|
|
|
# 品牌门店规模（按赛道聚合真实门店总数）
|
|
brand_stores = conn.execute(
|
|
"""
|
|
SELECT category, SUM(stores) AS total_stores, COUNT(*) AS brand_cnt
|
|
FROM brands WHERE category IS NOT NULL AND category != '' AND stores > 0
|
|
GROUP BY category
|
|
"""
|
|
).fetchall()
|
|
|
|
# 舆情分布（按赛道统计正面/负面/中性文章数）
|
|
sentiment_rows = conn.execute(
|
|
"""
|
|
SELECT a.category,
|
|
SUM(CASE WHEN json_extract(x.result_json, '$.sentiment') = 'positive' THEN 1 ELSE 0 END) AS positive_cnt,
|
|
SUM(CASE WHEN json_extract(x.result_json, '$.sentiment') = 'negative' THEN 1 ELSE 0 END) AS negative_cnt,
|
|
SUM(CASE WHEN json_extract(x.result_json, '$.sentiment') = 'neutral' OR json_extract(x.result_json, '$.sentiment') IS NULL THEN 1 ELSE 0 END) AS neutral_cnt
|
|
FROM articles a
|
|
JOIN ai_analyses x ON x.article_id = a.id AND x.analysis_type = 'editorial'
|
|
WHERE a.crawl_status = 'success' AND a.category IS NOT NULL AND a.category != ''
|
|
GROUP BY a.category
|
|
"""
|
|
).fetchall()
|
|
|
|
# 客单价分布（从品牌表聚合）
|
|
price_rows = conn.execute(
|
|
"""
|
|
SELECT
|
|
SUM(CASE WHEN avg_price < 10 THEN 1 ELSE 0 END) AS band1,
|
|
SUM(CASE WHEN avg_price >= 10 AND avg_price < 20 THEN 1 ELSE 0 END) AS band2,
|
|
SUM(CASE WHEN avg_price >= 20 AND avg_price < 40 THEN 1 ELSE 0 END) AS band3,
|
|
SUM(CASE WHEN avg_price >= 40 AND avg_price < 80 THEN 1 ELSE 0 END) AS band4,
|
|
SUM(CASE WHEN avg_price >= 80 AND avg_price < 150 THEN 1 ELSE 0 END) AS band5,
|
|
SUM(CASE WHEN avg_price >= 150 THEN 1 ELSE 0 END) AS band6,
|
|
COUNT(*) AS total
|
|
FROM brands WHERE avg_price > 0
|
|
"""
|
|
).fetchone()
|
|
|
|
# 热词热力图：近7天每天各标签出现次数
|
|
tag_heatmap = conn.execute(
|
|
"""
|
|
SELECT value AS tag,
|
|
strftime('%Y-%m-%d', a.published_at) AS day,
|
|
COUNT(*) AS cnt
|
|
FROM ai_analyses x
|
|
JOIN articles a ON a.id = x.article_id
|
|
, json_each(json_extract(x.result_json, '$.tags'))
|
|
WHERE a.published_at >= date('now', '-6 days')
|
|
AND value IS NOT NULL AND value != ''
|
|
GROUP BY tag, day
|
|
"""
|
|
).fetchall()
|
|
|
|
# 热词总量排行（用于选取 TOP 标签）
|
|
tag_total = conn.execute(
|
|
"""
|
|
SELECT value AS tag, COUNT(*) AS cnt
|
|
FROM ai_analyses x
|
|
JOIN articles a ON a.id = x.article_id
|
|
, json_each(json_extract(x.result_json, '$.tags'))
|
|
WHERE a.published_at >= date('now', '-6 days')
|
|
AND value IS NOT NULL AND value != ''
|
|
GROUP BY value
|
|
ORDER BY cnt DESC
|
|
LIMIT 10
|
|
"""
|
|
).fetchall()
|
|
|
|
# 构建热度趋势数据（热度 = 文章数 × 平均评分 / 10）
|
|
heat_trend = {}
|
|
for r in heat_rows:
|
|
month_key = f"{r['month'][-2:]}月" if r["month"] else "未知"
|
|
if month_key not in heat_trend:
|
|
heat_trend[month_key] = {"month": month_key}
|
|
heat_val = round((r["article_cnt"] or 0) * (r["avg_score"] or 0) / 10)
|
|
heat_trend[month_key][r["category"]] = heat_val
|
|
|
|
# 构建品牌门店规模数据
|
|
brand_stores_data = [
|
|
{"category": r["category"], "stores": r["total_stores"] or 0, "brands": r["brand_cnt"]}
|
|
for r in brand_stores
|
|
]
|
|
|
|
# 构建舆情分布数据
|
|
sentiment_data = [
|
|
{"category": r["category"], "正面": r["positive_cnt"] or 0, "负面": r["negative_cnt"] or 0, "中性": r["neutral_cnt"] or 0}
|
|
for r in sentiment_rows
|
|
]
|
|
|
|
# 构建价格分布
|
|
total_brands = price_rows["total"] or 1
|
|
price_bands = [
|
|
{"band": "10元以下", "pct": round((price_rows["band1"] or 0) * 100 / total_brands)},
|
|
{"band": "10-20元", "pct": round((price_rows["band2"] or 0) * 100 / total_brands)},
|
|
{"band": "20-40元", "pct": round((price_rows["band3"] or 0) * 100 / total_brands)},
|
|
{"band": "40-80元", "pct": round((price_rows["band4"] or 0) * 100 / total_brands)},
|
|
{"band": "80-150元", "pct": round((price_rows["band5"] or 0) * 100 / total_brands)},
|
|
{"band": "150元以上", "pct": round((price_rows["band6"] or 0) * 100 / total_brands)},
|
|
]
|
|
|
|
# 构建热词热力图数据（用 UTC 日期与 SQLite date('now') 保持一致）
|
|
from datetime import timedelta, timezone
|
|
today_utc = datetime.now(timezone.utc).date()
|
|
days = [(today_utc - timedelta(days=i)).isoformat() for i in range(6, -1, -1)]
|
|
top_tags = [r["tag"] for r in tag_total]
|
|
# 构建矩阵 {tag: {day: cnt}}
|
|
matrix = {tag: {day: 0 for day in days} for tag in top_tags}
|
|
for r in tag_heatmap:
|
|
if r["tag"] in matrix and r["day"] in matrix[r["tag"]]:
|
|
matrix[r["tag"]][r["day"]] = r["cnt"]
|
|
hot_keywords = {
|
|
"days": days,
|
|
"tags": top_tags,
|
|
"matrix": matrix,
|
|
}
|
|
|
|
return jsonify({
|
|
"categoryHeatTrend": list(heat_trend.values()),
|
|
"brandStoresByCategory": brand_stores_data,
|
|
"sentimentByCategory": sentiment_data,
|
|
"priceBandData": price_bands,
|
|
"hotKeywords": hot_keywords,
|
|
})
|
|
|
|
|
|
# ---- 报告 ----
|
|
|
|
@app.get("/api/reports")
|
|
def report_list():
|
|
with db() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM reports ORDER BY report_date DESC"
|
|
).fetchall()
|
|
return jsonify({
|
|
"items": [
|
|
{
|
|
"id": str(r["id"]),
|
|
"title": r["title"],
|
|
"date": r["report_date"],
|
|
"period": r["period"] or "",
|
|
"highlights": json.loads(r["highlights_json"]) if r["highlights_json"] else [],
|
|
"sections": json.loads(r["sections_json"]) if r["sections_json"] else [],
|
|
}
|
|
for r in rows
|
|
],
|
|
})
|
|
|
|
|
|
# ---- 报告生成 ----
|
|
|
|
@app.post("/api/reports/generate")
|
|
def report_generate():
|
|
days = (request.get_json(silent=True) or {}).get("days", 7)
|
|
try:
|
|
proc = subprocess.Popen(
|
|
[sys.executable, str(BASE_DIR / "structurer.py"), "report", "--days", str(days)],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
cwd=str(BASE_DIR),
|
|
)
|
|
stdout, stderr = proc.communicate(timeout=180)
|
|
if proc.returncode == 0:
|
|
output = stdout.decode("utf-8", errors="replace").strip()
|
|
try:
|
|
result = json.loads(output.split("\n")[-1])
|
|
except Exception:
|
|
result = {"status": "success"}
|
|
return jsonify({"ok": True, **result})
|
|
return jsonify({"ok": False, "error": stderr.decode("utf-8", errors="replace")[:500]}), 500
|
|
except subprocess.TimeoutExpired:
|
|
return jsonify({"ok": False, "error": "生成超时"}), 504
|
|
except Exception as exc:
|
|
return jsonify({"ok": False, "error": str(exc)}), 500
|
|
|
|
|
|
# ---- 行业信号 ----
|
|
|
|
@app.get("/api/signals")
|
|
def signal_list():
|
|
days = request.args.get("days", 30, type=int)
|
|
from datetime import datetime, timedelta
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
|
|
|
|
with db() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, signal_date, title, summary, trend, confidence,
|
|
implications_json, article_ids_json, tags_json, model
|
|
FROM industry_signals
|
|
WHERE signal_date >= ?
|
|
ORDER BY signal_date DESC, confidence DESC
|
|
""",
|
|
(cutoff,),
|
|
).fetchall()
|
|
|
|
# 批量查关联文章标题
|
|
all_ids = set()
|
|
for r in rows:
|
|
for aid in json.loads(r["article_ids_json"] or "[]"):
|
|
all_ids.add(aid)
|
|
article_map = {}
|
|
if all_ids:
|
|
placeholders = ",".join("?" * len(all_ids))
|
|
for ar in conn.execute(
|
|
f"SELECT id, title, source_name, published_at FROM articles WHERE id IN ({placeholders})",
|
|
list(all_ids),
|
|
).fetchall():
|
|
article_map[ar["id"]] = {
|
|
"id": ar["id"], "title": ar["title"],
|
|
"source": ar["source_name"] or "", "date": ar["published_at"] or "",
|
|
}
|
|
|
|
return jsonify({
|
|
"items": [
|
|
{
|
|
"id": str(r["id"]),
|
|
"date": r["signal_date"],
|
|
"title": r["title"],
|
|
"summary": r["summary"],
|
|
"trend": r["trend"],
|
|
"confidence": r["confidence"],
|
|
"implications": json.loads(r["implications_json"] or "[]"),
|
|
"articles": [
|
|
article_map[aid] for aid in json.loads(r["article_ids_json"] or "[]")
|
|
if aid in article_map
|
|
],
|
|
"tags": json.loads(r["tags_json"] or "[]"),
|
|
"model": r["model"],
|
|
}
|
|
for r in rows
|
|
],
|
|
})
|
|
|
|
|
|
# ---- RAG 问答 ----
|
|
|
|
@app.post("/api/qa/ask")
|
|
def qa_ask():
|
|
data = request.get_json(silent=True) or {}
|
|
question = (data.get("question") or "").strip()
|
|
if not question:
|
|
return jsonify({"ok": False, "error": "请输入问题"}), 400
|
|
|
|
try:
|
|
from rag import rag_qa
|
|
result = rag_qa(question)
|
|
return jsonify({"ok": True, **result})
|
|
except Exception as exc:
|
|
return jsonify({"ok": False, "error": str(exc)}), 502
|
|
|
|
|
|
@app.post("/api/qa/stream")
|
|
def qa_stream():
|
|
"""SSE 流式问答端点。"""
|
|
data = request.get_json(silent=True) or {}
|
|
question = (data.get("question") or "").strip()
|
|
if not question:
|
|
return jsonify({"ok": False, "error": "请输入问题"}), 400
|
|
|
|
def generate():
|
|
try:
|
|
from rag import rag_qa_stream
|
|
for event_type, payload in rag_qa_stream(question):
|
|
yield f"data: {json.dumps({'type': event_type, 'data': payload}, ensure_ascii=False)}\n\n"
|
|
yield "data: [DONE]\n\n"
|
|
except Exception as exc:
|
|
yield f"data: {json.dumps({'type': 'error', 'data': str(exc)}, ensure_ascii=False)}\n\n"
|
|
yield "data: [DONE]\n\n"
|
|
|
|
return Response(
|
|
stream_with_context(generate()),
|
|
mimetype="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"X-Accel-Buffering": "no",
|
|
"Connection": "keep-alive",
|
|
},
|
|
)
|
|
|
|
|
|
# ---- 触发采集 ----
|
|
|
|
@app.post("/api/crawl/run")
|
|
def crawl_run():
|
|
data = request.get_json(silent=True) or {}
|
|
source_type = data.get("source_type", "公众号")
|
|
try:
|
|
proc = subprocess.Popen(
|
|
[sys.executable, str(BASE_DIR / "crawler.py"), "--source-type", source_type],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
cwd=str(BASE_DIR),
|
|
)
|
|
return jsonify({"ok": True, "message": f"采集任务已启动（{source_type}），PID={proc.pid}"})
|
|
except Exception as exc:
|
|
return jsonify({"ok": False, "error": str(exc)}), 500
|
|
|
|
|
|
# ---- 触发结构化分析 ----
|
|
|
|
@app.post("/api/analyze/run")
|
|
def analyze_run():
|
|
data = request.get_json(silent=True) or {}
|
|
force = bool(data.get("force", False))
|
|
try:
|
|
cmd = [sys.executable, str(BASE_DIR / "structurer.py"), "analyze"]
|
|
if force:
|
|
cmd.append("--force")
|
|
proc = subprocess.Popen(
|
|
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
cwd=str(BASE_DIR),
|
|
)
|
|
return jsonify({"ok": True, "message": f"结构化分析任务已启动，PID={proc.pid}"})
|
|
except Exception as exc:
|
|
return jsonify({"ok": False, "error": str(exc)}), 500
|
|
|
|
|
|
# ---- 采集运行记录 ----
|
|
|
|
@app.get("/api/crawl/runs")
|
|
def crawl_runs():
|
|
with db() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM crawl_runs ORDER BY started_at DESC LIMIT 20"
|
|
).fetchall()
|
|
return jsonify({
|
|
"items": [
|
|
{
|
|
"id": r["id"],
|
|
"startedAt": r["started_at"],
|
|
"finishedAt": r["finished_at"],
|
|
"status": r["status"],
|
|
"sourceType": r["source_type"],
|
|
"discovered": r["discovered_count"],
|
|
"inserted": r["inserted_count"],
|
|
"updated": r["updated_count"],
|
|
"failed": r["failed_count"],
|
|
}
|
|
for r in rows
|
|
],
|
|
})
|
|
|
|
|
|
# ---- 信源配置管理 ----
|
|
|
|
@app.get("/api/sources")
|
|
def sources_list():
|
|
source_type = request.args.get("type", "")
|
|
with db() as conn:
|
|
if source_type:
|
|
rows = conn.execute(
|
|
"SELECT * FROM sources WHERE source_type=? ORDER BY enabled DESC, id DESC",
|
|
(source_type,),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT * FROM sources ORDER BY source_type, enabled DESC, id DESC"
|
|
).fetchall()
|
|
return jsonify({
|
|
"items": [
|
|
{
|
|
"id": r["id"],
|
|
"sourceType": r["source_type"],
|
|
"name": r["name"],
|
|
"finder": r["finder"] or "",
|
|
"listUrl": r["list_url"] or "",
|
|
"downloadDir": r["download_dir"] or "",
|
|
"enabled": bool(r["enabled"]),
|
|
"lastCrawledAt": r["last_crawled_at"],
|
|
}
|
|
for r in rows
|
|
],
|
|
})
|
|
|
|
|
|
@app.post("/api/sources")
|
|
def source_create():
|
|
data = request.get_json(silent=True) or {}
|
|
name = (data.get("name") or "").strip()
|
|
source_type = data.get("sourceType", "公众号")
|
|
if not name:
|
|
return jsonify({"ok": False, "error": "名称不能为空"}), 400
|
|
timestamp = now_iso()
|
|
with db() as conn:
|
|
try:
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT INTO sources (source_type, name, finder, list_url, download_dir, enabled, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
source_type, name,
|
|
data.get("finder", ""), data.get("listUrl", ""), data.get("downloadDir", ""),
|
|
int(data.get("enabled", True)), timestamp, timestamp,
|
|
),
|
|
)
|
|
conn.commit()
|
|
return jsonify({"ok": True, "id": cursor.lastrowid})
|
|
except sqlite3.IntegrityError:
|
|
return jsonify({"ok": False, "error": f"{source_type}「{name}」已存在"}), 409
|
|
|
|
|
|
@app.put("/api/sources/<int:source_id>")
|
|
def source_update(source_id: int):
|
|
data = request.get_json(silent=True) or {}
|
|
timestamp = now_iso()
|
|
with db() as conn:
|
|
row = conn.execute("SELECT * FROM sources WHERE id=?", (source_id,)).fetchone()
|
|
if not row:
|
|
return jsonify({"ok": False, "error": "信源不存在"}), 404
|
|
conn.execute(
|
|
"""
|
|
UPDATE sources SET
|
|
name=?, finder=?, list_url=?, download_dir=?, enabled=?, updated_at=?
|
|
WHERE id=?
|
|
""",
|
|
(
|
|
(data.get("name") or row["name"]).strip(),
|
|
data.get("finder", row["finder"]),
|
|
data.get("listUrl", row["list_url"]),
|
|
data.get("downloadDir", row["download_dir"]),
|
|
int(data.get("enabled", bool(row["enabled"]))),
|
|
timestamp, source_id,
|
|
),
|
|
)
|
|
conn.commit()
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.delete("/api/sources/<int:source_id>")
|
|
def source_delete(source_id: int):
|
|
with db() as conn:
|
|
row = conn.execute("SELECT * FROM sources WHERE id=?", (source_id,)).fetchone()
|
|
if not row:
|
|
return jsonify({"ok": False, "error": "信源不存在"}), 404
|
|
conn.execute("DELETE FROM sources WHERE id=?", (source_id,))
|
|
conn.commit()
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.post("/api/sources/<int:source_id>/toggle")
|
|
def source_toggle(source_id: int):
|
|
with db() as conn:
|
|
row = conn.execute("SELECT enabled FROM sources WHERE id=?", (source_id,)).fetchone()
|
|
if not row:
|
|
return jsonify({"ok": False, "error": "信源不存在"}), 404
|
|
new_val = 0 if row["enabled"] else 1
|
|
conn.execute("UPDATE sources SET enabled=?, updated_at=? WHERE id=?", (new_val, now_iso(), source_id))
|
|
conn.commit()
|
|
return jsonify({"ok": True, "enabled": bool(new_val)})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
ensure_schema()
|
|
port = int(os.getenv("PORT", "8765"))
|
|
app.run(host="127.0.0.1", port=port, debug=True)
|