Files
material-library/app.py
T

1146 lines
44 KiB
Python

# -*- coding: utf-8 -*-
"""
素材库系统 (Material Library)
- 项目: 分类/标签/描述管理
- 素材: 上传文件(文字/文档/图片/视频/音频) + 直接粘贴文本
- 文本抽取: txt/md/pdf/docx/pptx/csv/srt/json/xml/html
- 大模型分析: DeepSeek 提取摘要/要点/关键词/标签/归类; 图片走 doubao 视觉
- 索引: SQLite FTS5 (jieba 分词) 全文检索
"""
import os
import re
import io
import json
import uuid
import time
import zipfile
import shutil
import threading
import datetime
import sqlite3
from pathlib import Path
import requests
import jieba
from flask import Flask, request, jsonify, send_file, abort, g
# ---------------- 基础配置 ----------------
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
UPLOAD_DIR = DATA_DIR / "uploads"
DB_PATH = DATA_DIR / "material.db"
STATIC_DIR = BASE_DIR / "static"
BACKUP_DIR = DATA_DIR / "backups"
os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(STATIC_DIR, exist_ok=True)
os.makedirs(BACKUP_DIR, exist_ok=True)
MAX_FILE_MB = 500 # 单文件上限
MAX_TEXT_LEN = 30000 # 抽取文本送入 LLM 的最大长度
# ---------------- 大模型配置 ----------------
LLM_BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")
LLM_API_KEY = os.environ.get("LLM_API_KEY", "sk-edb9df58ff574f8c98df1cd6a425e97c")
LLM_MODEL = os.environ.get("LLM_MODEL", "deepseek-v4-flash")
VISION_BASE_URL = os.environ.get("VISION_BASE_URL", "https://ark.cn-beijing.volces.com/api/plan/v3")
VISION_API_KEY = os.environ.get("VISION_API_KEY", "ark-2b06dc9d-8878-4c6e-b201-f422376e79cb-246d9")
VISION_MODEL = os.environ.get("VISION_MODEL", "doubao-seed-evolving")
app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="/static")
app.config["MAX_CONTENT_LENGTH"] = MAX_FILE_MB * 1024 * 1024
_lock = threading.RLock()
# 允许的文件扩展名 -> 素材类型
TYPE_BY_EXT = {
".txt": "text", ".md": "text", ".markdown": "text", ".log": "text", ".srt": "text",
".csv": "text", ".json": "text", ".xml": "text", ".html": "text", ".htm": "text",
".pdf": "doc", ".docx": "doc", ".doc": "doc", ".pptx": "doc", ".ppt": "doc",
".jpg": "image", ".jpeg": "image", ".png": "image", ".gif": "image", ".bmp": "image",
".webp": "image", ".tif": "image", ".tiff": "image", ".svg": "image", ".ico": "image",
".mp4": "video", ".mov": "video", ".avi": "video", ".mkv": "video", ".flv": "video",
".wmv": "video", ".webm": "video", ".m4v": "video", ".ts": "video",
".mp3": "audio", ".wav": "audio", ".flac": "audio", ".aac": "audio", ".ogg": "audio",
".m4a": "audio", ".wma": "audio", ".amr": "audio",
}
TEXT_EXTS = {".txt", ".md", ".markdown", ".log", ".srt", ".csv", ".json", ".xml", ".html", ".htm"}
DOC_EXTS = {".pdf", ".docx", ".pptx"}
IMAGE_EXTS = {k for k, v in TYPE_BY_EXT.items() if v == "image"}
VIDEO_EXTS = {k for k, v in TYPE_BY_EXT.items() if v == "video"}
AUDIO_EXTS = {k for k, v in TYPE_BY_EXT.items() if v == "audio"}
ALLOWED_EXT = set(TYPE_BY_EXT.keys())
# ---------------- 数据库 ----------------
def get_db():
if "db" not in g:
g.db = sqlite3.connect(str(DB_PATH))
g.db.row_factory = sqlite3.Row
g.db.execute("PRAGMA journal_mode=WAL")
g.db.execute("PRAGMA foreign_keys=ON")
return g.db
def close_db(e=None):
db = g.pop("db", None)
if db is not None:
db.close()
def init_db():
db = sqlite3.connect(str(DB_PATH))
db.executescript("""
CREATE TABLE IF NOT EXISTS projects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
description TEXT DEFAULT '',
category TEXT DEFAULT '未分类',
tags TEXT DEFAULT '',
created_at TEXT DEFAULT (datetime('now','localtime')),
updated_at TEXT DEFAULT (datetime('now','localtime'))
);
CREATE TABLE IF NOT EXISTS materials (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER NOT NULL,
name TEXT NOT NULL,
mtype TEXT DEFAULT 'text',
original_name TEXT DEFAULT '',
file_path TEXT DEFAULT '',
file_size INTEGER DEFAULT 0,
ext TEXT DEFAULT '',
meta_json TEXT DEFAULT '{}',
extracted_text TEXT DEFAULT '',
text_status TEXT DEFAULT 'pending',
status TEXT DEFAULT 'uploaded',
created_at TEXT DEFAULT (datetime('now','localtime'))
);
CREATE TABLE IF NOT EXISTS analyses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
material_id INTEGER,
project_id INTEGER,
scope TEXT DEFAULT 'material',
summary TEXT DEFAULT '',
key_points TEXT DEFAULT '[]',
keywords TEXT DEFAULT '[]',
tags TEXT DEFAULT '[]',
category TEXT DEFAULT '',
detail TEXT DEFAULT '{}',
model TEXT DEFAULT '',
created_at TEXT DEFAULT (datetime('now','localtime'))
);
CREATE INDEX IF NOT EXISTS idx_mat_project ON materials(project_id);
CREATE INDEX IF NOT EXISTS idx_ana_project ON analyses(project_id);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT
);
CREATE VIRTUAL TABLE IF NOT EXISTS materials_fts USING fts5(term);
""")
db.commit()
# 写入默认设置(不覆盖已有)
defaults = {
"llm_base_url": LLM_BASE_URL, "llm_api_key": LLM_API_KEY, "llm_model": LLM_MODEL,
"vision_base_url": VISION_BASE_URL, "vision_api_key": VISION_API_KEY, "vision_model": VISION_MODEL,
"backup_interval_hours": "24", "backup_change_threshold": "50", "backup_max_keep": "10",
"last_backup_time": "", "change_counter": "0",
}
for k, v in defaults.items():
db.execute("INSERT OR IGNORE INTO settings(key,value) VALUES(?,?)", (k, v))
db.commit()
db.close()
def _ts():
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def _tokenize(text):
return " ".join(t for t in jieba.cut(text or "") if t.strip())
def _index_material(db, mid, name, text, summary="", keywords=""):
term = " ".join(filter(None, [_tokenize(name), _tokenize(text), _tokenize(summary), _tokenize(keywords)]))
db.execute("DELETE FROM materials_fts WHERE rowid=?", (mid,))
if term.strip():
db.execute("INSERT INTO materials_fts(rowid, term) VALUES (?,?)", (mid, term))
# ---------------- 文本抽取 ----------------
def _read_plain(path):
"""读取纯文本,尝试多种编码"""
raw = path.read_bytes()
for enc in ("utf-8", "utf-8-sig", "gb18030", "big5", "latin-1"):
try:
return raw.decode(enc)
except Exception:
continue
return raw.decode("utf-8", errors="ignore")
def _extract_docx(path):
import docx
d = docx.Document(str(path))
parts = [p.text for p in d.paragraphs if p.text.strip()]
for tbl in d.tables:
for row in tbl.rows:
cells = [c.text.strip() for c in row.cells if c.text.strip()]
if cells:
parts.append(" | ".join(cells))
return "\n".join(parts)
def _extract_pptx(path):
import zipfile
import xml.etree.ElementTree as ET
ns = {"a": "http://schemas.openxmlformats.org/drawingml/2006/main"}
parts = []
with zipfile.ZipFile(str(path)) as z:
slides = sorted([n for n in z.namelist() if re.match(r"ppt/slides/slide\d+\.xml$", n)],
key=lambda x: int(re.search(r"\d+", x.split("/")[-1]).group()))
for s in slides:
try:
root = ET.fromstring(z.read(s))
texts = [t.text for t in root.iter("{http://schemas.openxmlformats.org/drawingml/2006/main}t") if t.text and t.text.strip()]
if texts:
parts.append(" ".join(texts))
except Exception:
continue
return "\n".join(parts)
def _extract_pdf(path):
import pypdf
reader = pypdf.PdfReader(str(path))
parts = []
for i, page in enumerate(reader.pages[:200]):
try:
t = page.extract_text() or ""
if t.strip():
parts.append(t)
except Exception:
continue
return "\n".join(parts)
def extract_text(path, ext):
"""返回 (text, ok)"""
try:
if ext in (".pdf",):
t = _extract_pdf(path)
elif ext == ".docx":
t = _extract_docx(path)
elif ext == ".pptx":
t = _extract_pptx(path)
elif ext in TEXT_EXTS:
t = _read_plain(path)
else:
return "", False
t = re.sub(r"[ \t]+", " ", t)
t = re.sub(r"\n{3,}", "\n\n", t)
return t.strip(), bool(t.strip())
except Exception as e:
return f"<抽取失败: {e}>", False
def _meta_image(path):
from PIL import Image
try:
im = Image.open(str(path))
return {"width": im.width, "height": im.height, "format": im.format or ""}
except Exception:
return {}
def _safe_filename(name):
name = Path(name).name
name = re.sub(r"[\\/:*?\"<>|\s]+", "_", name)
return name[:120] or "unnamed"
# ---------------- 设置 & 备份机制 ----------------
def get_settings(db=None):
if db is None:
db = sqlite3.connect(str(DB_PATH))
db.row_factory = sqlite3.Row
close = True
else:
close = False
rows = db.execute("SELECT key,value FROM settings").fetchall()
s = {r["key"]: r["value"] for r in rows}
if close:
db.close()
return s
def get_setting(key, default=""):
s = get_settings()
return s.get(key) if s.get(key) not in (None, "") else default
def set_setting(db, key, value):
db.execute("INSERT INTO settings(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, str(value)))
def bump_changes(db):
"""变更计数+1(用于按变更量触发备份)"""
cur = db.execute("SELECT value FROM settings WHERE key='change_counter'").fetchone()
cnt = int(cur["value"]) if cur and cur["value"] else 0
set_setting(db, "change_counter", str(cnt + 1))
def llm_config():
return {
"base_url": get_setting("llm_base_url", LLM_BASE_URL),
"api_key": get_setting("llm_api_key", LLM_API_KEY),
"model": get_setting("llm_model", LLM_MODEL),
}
def vision_config():
return {
"base_url": get_setting("vision_base_url", VISION_BASE_URL),
"api_key": get_setting("vision_api_key", VISION_API_KEY),
"model": get_setting("vision_model", VISION_MODEL),
}
def make_backup(manual=False):
"""生成备份 zip(数据库 + 上传目录)"""
with _lock:
c = sqlite3.connect(str(DB_PATH))
try:
c.execute("PRAGMA wal_checkpoint(TRUNCATE)")
finally:
c.close()
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
name = f"backup_{ts}{'_manual' if manual else ''}.zip"
zpath = BACKUP_DIR / name
with zipfile.ZipFile(zpath, "w", zipfile.ZIP_DEFLATED) as z:
z.write(DB_PATH, "data/material.db")
if UPLOAD_DIR.exists():
for p in sorted(UPLOAD_DIR.rglob("*")):
if p.is_file():
z.write(p, f"data/uploads/{p.relative_to(UPLOAD_DIR)}")
db = sqlite3.connect(str(DB_PATH))
set_setting(db, "last_backup_time", _ts())
set_setting(db, "change_counter", "0")
db.commit()
db.close()
_prune_backups()
return name
def _prune_backups():
try:
max_keep = int(get_setting("backup_max_keep", "10") or 10)
except Exception:
max_keep = 10
files = sorted(BACKUP_DIR.glob("backup_*.zip"), key=lambda p: p.stat().st_mtime, reverse=True)
for f in files[max_keep:]:
try:
f.unlink()
except Exception:
pass
def list_backups():
out = []
for p in sorted(BACKUP_DIR.glob("backup_*.zip"), key=lambda x: x.stat().st_mtime, reverse=True):
out.append({"name": p.name, "size": p.stat().st_size,
"time": datetime.datetime.fromtimestamp(p.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S")})
return out
def _backup_scheduler():
"""后台调度:按时间间隔 / 变更量 自动备份"""
while True:
try:
time.sleep(30)
s = get_settings()
try:
interval = float(s.get("backup_interval_hours", "24") or 0)
except Exception:
interval = 24
try:
threshold = int(s.get("backup_change_threshold", "50") or 0)
except Exception:
threshold = 50
try:
cnt = int(s.get("change_counter", "0") or 0)
except Exception:
cnt = 0
last = s.get("last_backup_time", "")
due_time = False
if interval > 0:
if not last:
due_time = True
else:
try:
lt = datetime.datetime.strptime(last, "%Y-%m-%d %H:%M:%S")
due_time = (datetime.datetime.now() - lt).total_seconds() >= interval * 3600
except Exception:
due_time = True
if (due_time or (threshold > 0 and cnt >= threshold)):
make_backup()
except Exception:
pass
# ---------------- 大模型调用 ----------------
def _llm_chat(messages, model=None, temperature=0.3, max_tokens=1500):
cfg = llm_config()
url = f"{cfg['base_url'].rstrip('/')}/chat/completions"
headers = {"Authorization": f"Bearer {cfg['api_key']}"}
payload_base = {"model": model or cfg["model"], "messages": messages,
"temperature": temperature, "max_tokens": max_tokens}
# 部分模型不支持 response_format=json_object,失败时自动降级重试
for use_json in (True, False):
payload = dict(payload_base)
if use_json:
payload["response_format"] = {"type": "json_object"}
try:
r = requests.post(url, headers=headers, json=payload, timeout=120)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
except requests.HTTPError:
if not use_json:
raise
continue
except Exception:
raise
raise RuntimeError("LLM 调用失败")
ANALYZE_PROMPT = """你是专业素材库分析助手。请对下面这个素材做深度分析,提取关键信息。
只输出合法 JSON,不要任何额外文字,格式如下:
{{"summary":"120字以内中文摘要","key_points":["要点1","要点2"...最多6条],"keywords":["关键词1"...最多8个],"tags":["标签1"...最多5个],"category":"建议归类,从[技术文档,产品资料,研究报告,市场资讯,项目管理,图片素材,音视频素材,其他]中选择"}}
素材名称:{name}
素材类型:{mtype}
{extra}
素材内容:
{content}
"""
def _parse_json(content):
try:
return json.loads(content)
except Exception:
m = re.search(r"\{.*\}", content, re.S)
if m:
try:
return json.loads(m.group())
except Exception:
pass
return {}
def _vision_describe(path):
"""用视觉大模型描述图片"""
try:
import base64
cfg = vision_config()
b64 = base64.b64encode(Path(path).read_bytes()).decode()
url = f"{cfg['base_url'].rstrip('/')}/chat/completions"
payload = {
"model": cfg["model"],
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": "请用中文简洁描述这张图片的内容(80字以内),说明主体、场景、画面中的文字信息(如有)。只输出描述本身。"},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
],
}],
"temperature": 0.3,
"max_tokens": 300,
}
r = requests.post(url, headers={"Authorization": f"Bearer {cfg['api_key']}"},
json=payload, timeout=120)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"].strip()
except Exception:
return None
def analyze_material(mat):
"""分析单个素材,返回 (analysis_dict, model, detail)"""
if not isinstance(mat, dict):
mat = dict(mat)
detail = {}
mtype = mat["mtype"]
content = mat.get("extracted_text") or ""
extra = ""
if mtype == "image" and mat.get("file_path"):
desc = _vision_describe(mat["file_path"])
if desc:
extra = "图片视觉描述:\n" + desc
detail["vision"] = desc
if mtype == "video":
extra = "视频素材(未转写音频),基于文件名与元数据分析"
if mtype == "audio":
extra = "音频素材(未转写语音),基于文件名与元数据分析"
if mtype == "other" or not content:
content = content or "(该素材无可提取文本内容)"
text = content[:MAX_TEXT_LEN]
prompt = ANALYZE_PROMPT.format(name=mat["name"], mtype=mtype, extra=extra, content=text)
out = _llm_chat([{"role": "user", "content": prompt}])
data = _parse_json(out)
model = llm_config()["model"]
detail["raw"] = out[:3000]
detail["model"] = model
return data, model, detail
def _store_analysis(db, material_id, project_id, scope, data, model, detail):
if scope == "material":
db.execute("DELETE FROM analyses WHERE material_id=? AND scope='material'", (material_id,))
# project 摘要:保留全部历史记录,不做删除
cur = db.execute(
"INSERT INTO analyses (material_id, project_id, scope, summary, key_points, keywords, tags, category, detail, model, created_at) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(material_id, project_id, scope,
(data.get("summary") or ""),
json.dumps(data.get("key_points", []), ensure_ascii=False),
json.dumps(data.get("keywords", []), ensure_ascii=False),
json.dumps(data.get("tags", []), ensure_ascii=False),
(data.get("category") or ""),
json.dumps(detail, ensure_ascii=False),
model, _ts()))
db.commit()
return cur.lastrowid
# ---------------- 后台任务 ----------------
_tasks = {}
def _task_key(scope, obj_id):
return f"{scope}:{obj_id}"
def _run_analysis(project_id, material_ids=None, scope="material"):
key = _task_key(scope, project_id)
_tasks[key] = {"running": True, "done": 0, "total": 0, "msg": "准备中", "error": None}
db = sqlite3.connect(str(DB_PATH))
db.row_factory = sqlite3.Row
try:
if scope == "project":
cond = "project_id=? AND (status='uploaded' OR status='error' OR status='analyzed')"
mats = db.execute(f"SELECT * FROM materials WHERE {cond}", (project_id,)).fetchall()
else:
if material_ids:
ph = ",".join("?" for _ in material_ids)
mats = db.execute(f"SELECT * FROM materials WHERE project_id=? AND id IN ({ph})", (project_id, *material_ids)).fetchall()
else:
# 未指定 id → 分析项目下所有待分析/失败素材
mats = db.execute(
"SELECT * FROM materials WHERE project_id=? AND (status='uploaded' OR status='error' OR status='analyzed')",
(project_id,)).fetchall()
_tasks[key]["total"] = len(mats)
for i, mat in enumerate(mats, 1):
mid = mat["id"]
try:
db.execute("UPDATE materials SET status='analyzing' WHERE id=?", (mid,))
db.commit()
data, model, detail = analyze_material(mat)
aid = _store_analysis(db, mid, project_id, "material", data, model, detail)
# 回写关键词作为标签候选 + 更新索引
keywords = data.get("keywords", [])
kw = ",".join(keywords[:8]) if isinstance(keywords, list) else ""
db.execute("UPDATE materials SET status='analyzed', name=name WHERE id=?", (mid,))
row = db.execute("SELECT name, extracted_text, summary FROM materials LEFT JOIN analyses ON analyses.id=? WHERE materials.id=?", (aid, mid)).fetchone()
_index_material(db, mid, row["name"], row["extracted_text"] or "", row["summary"] or "", kw)
db.commit()
except Exception as e:
db.execute("UPDATE materials SET status='error' WHERE id=?", (mid,))
db.commit()
_tasks[key]["error"] = f"{mat['name']}: {e}"
_tasks[key]["done"] = i
_tasks[key]["msg"] = f"已分析 {i}/{len(mats)}"
except Exception as e:
_tasks[key]["error"] = str(e)
finally:
_tasks[key]["running"] = False
if not _tasks[key].get("error"):
_tasks[key]["msg"] = "完成"
db.close()
def _run_project_summary(project_id):
key = _task_key("summary", project_id)
_tasks[key] = {"running": True, "done": 0, "total": 1, "msg": "生成中", "error": None}
db = sqlite3.connect(str(DB_PATH))
db.row_factory = sqlite3.Row
try:
proj = db.execute("SELECT * FROM projects WHERE id=?", (project_id,)).fetchone()
mats = db.execute("SELECT * FROM materials WHERE project_id=?", (project_id,)).fetchall()
# 汇总:已有素材级分析的摘要 + 原始文本
parts = []
for m in mats[:15]:
ana = db.execute("SELECT * FROM analyses WHERE material_id=? AND scope='material' ORDER BY id DESC LIMIT 1", (m["id"],)).fetchone()
if ana and ana["summary"]:
parts.append(f"- {m['name']}: {ana['summary']}")
else:
t = (m["extracted_text"] or "")[:300]
parts.append(f"- {m['name']}: {t}")
content = "\n".join(parts) or "(项目暂无素材)"
prompt = f"""你是项目管理助理。基于以下素材库项目的资料汇总,生成项目级分析报告。只输出合法 JSON:
{{"summary":"整个项目内容的总摘要(150字以内)","key_points":["项目关键发现"...最多8条],"keywords":["项目级关键词"...最多10个],"tags":["项目标签"...最多5个],"category":"建议项目类别"}}
项目名称:{proj['name']}
项目描述:{proj['description'] or '无'}
素材数量:{len(mats)}
资料汇总:
{content[:MAX_TEXT_LEN]}
"""
out = _llm_chat([{"role": "user", "content": prompt}])
data = _parse_json(out)
model = llm_config()["model"]
_store_analysis(db, None, project_id, "project", data, model, {"model": model, "raw": out[:3000]})
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), project_id))
db.commit()
_tasks[key]["done"] = 1
_tasks[key]["msg"] = "完成"
except Exception as e:
_tasks[key]["error"] = str(e)
finally:
_tasks[key]["running"] = False
db.close()
# ---------------- 辅助 ----------------
def project_json(row, db):
n = db.execute("SELECT COUNT(*) FROM materials WHERE project_id=?", (row["id"],)).fetchone()[0]
na = db.execute("SELECT COUNT(*) FROM materials WHERE project_id=? AND status='analyzed'", (row["id"],)).fetchone()[0]
ana = db.execute("SELECT * FROM analyses WHERE project_id=? AND scope='project' ORDER BY id DESC LIMIT 1", (row["id"],)).fetchone()
ana_d = None
if ana:
ana_d = dict(ana)
ana_d["key_points"] = json.loads(ana_d["key_points"] or "[]")
ana_d["keywords"] = json.loads(ana_d["keywords"] or "[]")
ana_d["tags"] = json.loads(ana_d["tags"] or "[]")
ana_d["detail"] = json.loads(ana_d["detail"] or "{}")
return {
"id": row["id"], "name": row["name"], "description": row["description"],
"category": row["category"], "tags": row["tags"], "created_at": row["created_at"],
"updated_at": row["updated_at"], "material_count": n, "analyzed_count": na,
"analysis": ana_d,
}
def material_json(row, db):
d = dict(row)
d["meta"] = json.loads(row["meta_json"] or "{}")
ana = db.execute("SELECT * FROM analyses WHERE material_id=? AND scope='material' ORDER BY id DESC LIMIT 1", (row["id"],)).fetchone()
d["analysis"] = dict(ana) if ana else None
if d["analysis"]:
d["analysis"]["key_points"] = json.loads(d["analysis"]["key_points"] or "[]")
d["analysis"]["keywords"] = json.loads(d["analysis"]["keywords"] or "[]")
d["analysis"]["tags"] = json.loads(d["analysis"]["tags"] or "[]")
d["analysis"]["detail"] = json.loads(d["analysis"]["detail"] or "{}")
return d
# ---------------- 路由:页面 ----------------
@app.route("/")
def index():
return send_file(str(STATIC_DIR / "index.html"))
@app.route("/files/<int:mid>")
def serve_file(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row or not row["file_path"]:
abort(404)
path = Path(row["file_path"])
if not path.exists():
abort(404)
ext = (row["ext"] or "").lower()
if ext in IMAGE_EXTS:
return send_file(str(path), mimetype="image/*", download_name=row["name"])
if ext in VIDEO_EXTS:
return send_file(str(path), mimetype="video/mp4", download_name=row["name"])
if ext in AUDIO_EXTS:
return send_file(str(path), mimetype="audio/mpeg", download_name=row["name"])
return send_file(str(path), as_attachment=True, download_name=row["original_name"] or row["name"])
# ---------------- 路由:项目 ----------------
@app.get("/api/projects")
def list_projects():
db = get_db()
q = (request.args.get("q") or "").strip()
category = (request.args.get("category") or "").strip()
tags_param = (request.args.get("tags") or request.args.get("tag") or "").strip()
sql = "SELECT * FROM projects WHERE 1=1"
args = []
if q:
sql += " AND (name LIKE ? OR description LIKE ? OR tags LIKE ?)"
like = f"%{q}%"
args += [like, like, like]
if category:
sql += " AND category=?"
args.append(category)
if tags_param:
# 多选标签:项目需同时包含所有选中标签(AND)
for t in [x.strip() for x in tags_param.split(",") if x.strip()]:
sql += " AND tags LIKE ?"
args.append(f"%{t}%")
sql += " ORDER BY updated_at DESC, id DESC"
rows = db.execute(sql, args).fetchall()
return jsonify([project_json(r, db) for r in rows])
@app.post("/api/projects")
def create_project():
d = request.get_json(force=True, silent=True) or {}
name = (d.get("name") or "").strip()
if not name:
return jsonify({"error": "项目名称不能为空"}), 400
db = get_db()
cur = db.execute("INSERT INTO projects (name, description, category, tags, updated_at) VALUES (?,?,?,?,?)",
(name, (d.get("description") or "").strip(),
(d.get("category") or "未分类").strip(),
(d.get("tags") or "").strip(), _ts()))
db.commit()
bump_changes(db)
db.commit()
row = db.execute("SELECT * FROM projects WHERE id=?", (cur.lastrowid,)).fetchone()
return jsonify(project_json(row, db))
@app.get("/api/projects/<int:pid>")
def get_project(pid):
db = get_db()
row = db.execute("SELECT * FROM projects WHERE id=?", (pid,)).fetchone()
if not row:
return jsonify({"error": "项目不存在"}), 404
p = project_json(row, db)
mats = db.execute("SELECT * FROM materials WHERE project_id=? ORDER BY id DESC", (pid,)).fetchall()
p["materials"] = [material_json(m, db) for m in mats]
return jsonify(p)
@app.put("/api/projects/<int:pid>")
def update_project(pid):
d = request.get_json(force=True, silent=True) or {}
db = get_db()
row = db.execute("SELECT * FROM projects WHERE id=?", (pid,)).fetchone()
if not row:
return jsonify({"error": "项目不存在"}), 404
name = (d.get("name") or row["name"]).strip()
db.execute("UPDATE projects SET name=?, description=?, category=?, tags=?, updated_at=? WHERE id=?",
(name, (d.get("description", row["description"]) or "").strip(),
(d.get("category", row["category"]) or "未分类").strip(),
(d.get("tags", row["tags"]) or "").strip(), _ts(), pid))
bump_changes(db)
db.commit()
return jsonify(project_json(db.execute("SELECT * FROM projects WHERE id=?", (pid,)).fetchone(), db))
@app.delete("/api/projects/<int:pid>")
def delete_project(pid):
db = get_db()
mats = db.execute("SELECT * FROM materials WHERE project_id=?", (pid,)).fetchall()
for m in mats:
if m["file_path"]:
p = Path(m["file_path"])
if p.exists():
try:
p.unlink()
except Exception:
pass
db.execute("DELETE FROM materials_fts WHERE rowid=?", (m["id"],))
db.execute("DELETE FROM analyses WHERE project_id=?", (pid,))
db.execute("DELETE FROM materials WHERE project_id=?", (pid,))
db.execute("DELETE FROM projects WHERE id=?", (pid,))
bump_changes(db)
db.commit()
return jsonify({"ok": True})
@app.get("/api/meta")
def get_meta():
db = get_db()
cats = db.execute("SELECT category, COUNT(*) c FROM projects GROUP BY category ORDER BY c DESC").fetchall()
all_tags = [r["tags"] for r in db.execute("SELECT tags FROM projects").fetchall() if r["tags"]]
tag_counter = {}
for t in all_tags:
for x in [x.strip() for x in t.split(",") if x.strip()]:
tag_counter[x] = tag_counter.get(x, 0) + 1
stats = {
"projects": db.execute("SELECT COUNT(*) FROM projects").fetchone()[0],
"materials": db.execute("SELECT COUNT(*) FROM materials").fetchone()[0],
"analyzed": db.execute("SELECT COUNT(*) FROM materials WHERE status='analyzed'").fetchone()[0],
}
return jsonify({
"categories": [{"name": r["category"], "count": r["c"]} for r in cats],
"tags": [{"name": k, "count": v} for k, v in sorted(tag_counter.items(), key=lambda x: -x[1])[:40]],
"stats": stats,
})
# ---------------- 路由:素材 ----------------
@app.post("/api/projects/<int:pid>/materials/upload")
def upload_materials(pid):
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
files = request.files.getlist("files")
results = []
for f in files:
if not f or not f.filename:
continue
orig = f.filename
ext = os.path.splitext(orig)[1].lower()
safe = _safe_filename(orig)
if ext not in ALLOWED_EXT:
results.append({"name": orig, "ok": False, "error": f"不支持的文件类型 {ext or '(无扩展名)'}"})
continue
pid_dir = UPLOAD_DIR / str(pid)
pid_dir.mkdir(parents=True, exist_ok=True)
uniq = uuid.uuid4().hex[:10]
dest = pid_dir / f"{uniq}_{safe}"
f.save(str(dest))
mtype = TYPE_BY_EXT.get(ext, "other")
# 提取文本
text, ok = extract_text(dest, ext)
text_status = "extracted" if ok else ("empty" if not text else "error")
# 元数据
meta = {}
if mtype == "image":
meta = _meta_image(dest)
size = dest.stat().st_size
cur = db.execute(
"INSERT INTO materials (project_id, name, mtype, original_name, file_path, file_size, ext, meta_json, extracted_text, text_status) "
"VALUES (?,?,?,?,?,?,?,?,?,?)",
(pid, safe, mtype, orig, str(dest), size, ext, json.dumps(meta, ensure_ascii=False), text, text_status))
mid = cur.lastrowid
_index_material(db, mid, safe, text)
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), pid))
bump_changes(db)
db.commit()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
results.append(material_json(row, db))
return jsonify({"results": results, "count": len(results)})
@app.post("/api/projects/<int:pid>/materials")
def add_text_material(pid):
d = request.get_json(force=True, silent=True) or {}
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
name = (d.get("name") or "").strip() or "未命名文本"
content = (d.get("content") or "")
cur = db.execute(
"INSERT INTO materials (project_id, name, mtype, original_name, file_size, meta_json, extracted_text, text_status, status) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(pid, name, "text", "", len(content.encode("utf-8")), "{}", content,
"extracted" if content.strip() else "empty", "uploaded"))
mid = cur.lastrowid
_index_material(db, mid, name, content)
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), pid))
bump_changes(db)
db.commit()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
return jsonify(material_json(row, db))
@app.get("/api/materials/<int:mid>")
def get_material(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
return jsonify(material_json(row, db))
@app.put("/api/materials/<int:mid>")
def update_material(mid):
d = request.get_json(force=True, silent=True) or {}
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
name = (d.get("name") or row["name"]).strip()
mtype = (d.get("mtype") or row["mtype"]).strip()
# 支持编辑文本内容:内容变了则旧分析作废,需重新分析
if "content" in d:
new_content = d.get("content") or ""
if new_content != (row["extracted_text"] or ""):
db.execute("UPDATE materials SET extracted_text=?, text_status=?, status='uploaded' WHERE id=?",
(new_content, "extracted" if new_content.strip() else "empty", mid))
db.execute("DELETE FROM analyses WHERE material_id=? AND scope='material'", (mid,))
db.execute("UPDATE materials SET name=?, mtype=? WHERE id=?", (name, mtype, mid))
# 更新索引
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
ana = db.execute("SELECT * FROM analyses WHERE material_id=? AND scope='material' ORDER BY id DESC LIMIT 1", (mid,)).fetchone()
kw = ",".join(json.loads(ana["keywords"] or "[]")) if ana else ""
_index_material(db, mid, name, row["extracted_text"] or "", (ana["summary"] if ana else "") or "", kw)
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), row["project_id"]))
bump_changes(db)
db.commit()
row2 = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
return jsonify(material_json(row2, db))
@app.delete("/api/materials/<int:mid>")
def delete_material(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
if row["file_path"]:
p = Path(row["file_path"])
if p.exists():
try:
p.unlink()
except Exception:
pass
db.execute("DELETE FROM materials_fts WHERE rowid=?", (mid,))
db.execute("DELETE FROM analyses WHERE material_id=?", (mid,))
db.execute("DELETE FROM materials WHERE id=?", (mid,))
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), row["project_id"]))
bump_changes(db)
db.commit()
return jsonify({"ok": True})
# ---------------- 路由:分析 ----------------
@app.post("/api/materials/<int:mid>/analyze")
def analyze_one(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
t = threading.Thread(target=_run_analysis, args=(row["project_id"], [mid]), daemon=True)
t.start()
return jsonify({"ok": True, "task": _task_key("material", row["project_id"])})
@app.post("/api/projects/<int:pid>/analyze")
def analyze_project(pid):
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
key = _task_key("material", pid)
t = threading.Thread(target=_run_analysis, args=(pid, None, "material"), daemon=True)
t.start()
return jsonify({"ok": True, "task": key})
@app.post("/api/projects/<int:pid>/summary")
def project_summary(pid):
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
t = threading.Thread(target=_run_project_summary, args=(pid,), daemon=True)
t.start()
return jsonify({"ok": True, "task": _task_key("summary", pid)})
@app.get("/api/tasks/<path:key>")
def task_status(key):
t = _tasks.get(key)
if not t:
return jsonify({"running": False, "done": 0, "total": 0, "msg": "无任务", "error": None})
return jsonify(t)
# ---------------- 路由:搜索 ----------------
@app.get("/api/search")
def search():
q = (request.args.get("q") or "").strip()
if not q:
return jsonify({"projects": [], "materials": []})
db = get_db()
# 1) 项目 LIKE
projs = db.execute("SELECT * FROM projects WHERE name LIKE ? OR description LIKE ? OR tags LIKE ? ORDER BY updated_at DESC LIMIT 10",
(f"%{q}%", f"%{q}%", f"%{q}%")).fetchall()
# 2) FTS 素材
materials = []
tokens = [t for t in jieba.cut(q) if t.strip()]
if tokens:
match = " AND ".join('"' + t + '"' for t in tokens[:6])
try:
ids = [r[0] for r in db.execute("SELECT rowid FROM materials_fts WHERE materials_fts MATCH ? ORDER BY rank LIMIT 50", (match,)).fetchall()]
if ids:
ph = ",".join("?" for _ in ids)
rows = db.execute(f"SELECT * FROM materials WHERE id IN ({ph})", ids).fetchall()
by_id = {r["id"]: r for r in rows}
for i in ids:
if i in by_id:
materials.append(by_id[i])
except Exception:
materials = []
# 3) LIKE 兜底
if not materials:
materials = db.execute("SELECT * FROM materials WHERE name LIKE ? OR extracted_text LIKE ? ORDER BY id DESC LIMIT 30",
(f"%{q}%", f"%{q}%")).fetchall()
return jsonify({
"projects": [project_json(p, db) for p in projs],
"materials": [material_json(m, db) for m in materials],
})
# ---------------- 路由:设置 & 备份 ----------------
@app.get("/api/settings")
def api_get_settings():
return jsonify(get_settings())
@app.put("/api/settings")
def api_put_settings():
d = request.get_json(force=True, silent=True) or {}
db = get_db()
keys = ["llm_base_url", "llm_api_key", "llm_model",
"vision_base_url", "vision_api_key", "vision_model",
"backup_interval_hours", "backup_change_threshold", "backup_max_keep"]
for k in keys:
if k in d:
set_setting(db, k, d[k])
db.commit()
return jsonify(get_settings())
@app.post("/api/settings/test")
def api_test_settings():
d = request.get_json(force=True, silent=True) or {}
kind = d.get("kind", "llm")
base_url = (d.get("base_url") or "").strip()
api_key = (d.get("api_key") or "").strip()
model = (d.get("model") or "").strip()
if not base_url or not api_key or not model:
return jsonify({"ok": False, "error": "请填写完整的接口地址 / Key / 模型名"}), 400
try:
payload = {"model": model, "messages": [{"role": "user", "content": "请用 json 格式回复:{\"status\":\"ok\"}"}], "max_tokens": 30}
if kind == "llm":
payload["response_format"] = {"type": "json_object"}
r = requests.post(f"{base_url.rstrip('/')}/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json=payload, timeout=60)
r.raise_for_status()
msg = r.json()["choices"][0]["message"]
reply = msg.get("content") or ""
if not reply.strip() and msg.get("reasoning_content"):
reply = "(推理过程) " + msg["reasoning_content"]
return jsonify({"ok": True, "reply": str(reply)[:200]})
except Exception as e:
return jsonify({"ok": False, "error": str(e)[:300]})
@app.get("/api/backups")
def api_list_backups():
return jsonify({"backups": list_backups(),
"last_backup_time": get_setting("last_backup_time", ""),
"change_counter": get_setting("change_counter", "0")})
@app.post("/api/backup")
def api_make_backup():
try:
name = make_backup(manual=True)
return jsonify({"ok": True, "name": name})
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 500
@app.get("/api/backup/download/<path:name>")
def api_download_backup(name):
safe = Path(name).name
if not re.fullmatch(r"backup_\d{8}_\d{6}(_manual)?\.zip", safe):
abort(400)
p = BACKUP_DIR / safe
if not p.exists():
abort(404)
return send_file(str(p), as_attachment=True, download_name=safe)
@app.delete("/api/backup/<path:name>")
def api_delete_backup(name):
safe = Path(name).name
if not re.fullmatch(r"backup_\d{8}_\d{6}(_manual)?\.zip", safe):
abort(400)
p = BACKUP_DIR / safe
if p.exists():
p.unlink()
return jsonify({"ok": True})
@app.post("/api/backup/restore")
def api_restore_backup():
for k, t in _tasks.items():
if t.get("running"):
return jsonify({"ok": False, "error": "当前有后台任务运行中,请稍后再恢复"}), 409
f = request.files.get("file")
if not f or not f.filename:
return jsonify({"ok": False, "error": "请选择备份文件"}), 400
tmp = DATA_DIR / "_restore_tmp"
if tmp.exists():
shutil.rmtree(str(tmp))
tmp.mkdir()
try:
with zipfile.ZipFile(f.stream) as z:
if "data/material.db" not in z.namelist():
shutil.rmtree(str(tmp))
return jsonify({"ok": False, "error": "备份文件不包含数据库"}), 400
z.extractall(str(tmp))
except Exception as e:
shutil.rmtree(str(tmp))
return jsonify({"ok": False, "error": f"备份文件解析失败: {e}"}), 400
try:
c = sqlite3.connect(str(DB_PATH))
c.execute("PRAGMA wal_checkpoint(TRUNCATE)")
c.close()
db_tmp = tmp / "data" / "material.db"
shutil.copy2(str(db_tmp), str(DB_PATH))
for suffix in ("-wal", "-shm"):
p = Path(str(DB_PATH) + suffix)
if p.exists():
p.unlink()
if UPLOAD_DIR.exists():
shutil.rmtree(str(UPLOAD_DIR))
up_tmp = tmp / "data" / "uploads"
if up_tmp.exists():
shutil.move(str(up_tmp), str(UPLOAD_DIR))
else:
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
init_db()
shutil.rmtree(str(tmp))
return jsonify({"ok": True, "msg": "恢复成功"})
except Exception as e:
shutil.rmtree(str(tmp))
return jsonify({"ok": False, "error": f"恢复失败: {e}"}), 500
# ---------------- 路由:摘要历史 ----------------
@app.get("/api/projects/<int:pid>/summaries")
def project_summaries(pid):
db = get_db()
rows = db.execute("SELECT * FROM analyses WHERE project_id=? AND scope='project' ORDER BY id DESC", (pid,)).fetchall()
out = []
for r in rows:
d = dict(r)
d["key_points"] = json.loads(d["key_points"] or "[]")
d["keywords"] = json.loads(d["keywords"] or "[]")
d["tags"] = json.loads(d["tags"] or "[]")
d["detail"] = json.loads(d["detail"] or "{}")
out.append(d)
return jsonify(out)
app.teardown_appcontext(close_db)
if __name__ == "__main__":
init_db()
threading.Thread(target=_backup_scheduler, daemon=True).start()
port = int(os.environ.get("PORT", "16091"))
print(f"素材库系统启动: http://0.0.0.0:{port}")
app.run(host="0.0.0.0", port=port, debug=False, threaded=True)