407 lines
15 KiB
Python
407 lines
15 KiB
Python
"""PMFlow FastAPI 应用入口。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import sqlalchemy as sa
|
|
|
|
from .api.router import api_router
|
|
from .core.config import settings
|
|
from .core.logging import configure_logging, get_logger
|
|
from .core.response import success_response
|
|
from .database import AsyncSessionLocal, engine
|
|
from .models import Base
|
|
from .services.scheduler_center import WorkloadRecalcScheduler
|
|
from .services.scheduler_center import KnowledgeUrlRefreshScheduler
|
|
from .services.scheduler_center import KnowledgeGraphIncrementalScheduler
|
|
from .services.scheduler_center import AIPMOCadenceScheduler
|
|
from .services.ai_orchestration_run_service import AIOrchestrationRunService
|
|
from .services.skill_service import init_system_skill_tags
|
|
from .services.feishu_long_connection import init_feishu_long_connection_manager
|
|
from .services.feishu_integration import dispatch_pending_feishu_replies
|
|
from .services.context_cache import register_context_cache_invalidation_hooks
|
|
|
|
configure_logging(settings.debug)
|
|
logger = get_logger(__name__)
|
|
register_context_cache_invalidation_hooks()
|
|
|
|
|
|
def _sqlite_backfill_columns(sync_conn) -> None:
|
|
"""为历史 SQLite 库补齐新增列，避免运行时 500。"""
|
|
dialect_name = str(sync_conn.dialect.name or "").lower()
|
|
if dialect_name != "sqlite":
|
|
return
|
|
|
|
def _columns(table_name: str) -> set[str]:
|
|
try:
|
|
rows = sync_conn.exec_driver_sql(f"PRAGMA table_info({table_name})").fetchall()
|
|
return {str(row[1]) for row in rows}
|
|
except Exception:
|
|
return set()
|
|
|
|
def _add_column(table_name: str, column_name: str, column_ddl: str) -> None:
|
|
cols = _columns(table_name)
|
|
if not cols or column_name in cols:
|
|
return
|
|
sync_conn.exec_driver_sql(
|
|
f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_ddl}"
|
|
)
|
|
|
|
# truth_nodes
|
|
_add_column("truth_nodes", "signal_id", "BIGINT")
|
|
|
|
# signals
|
|
_add_column("signals", "feishu_item_id", "BIGINT")
|
|
_add_column("signals", "matched_project_id", "BIGINT")
|
|
_add_column("signals", "title", "VARCHAR(200)")
|
|
_add_column("signals", "signal_type", "VARCHAR(50)")
|
|
_add_column("signals", "summary", "TEXT")
|
|
_add_column("signals", "reasoning", "TEXT")
|
|
_add_column("signals", "confidence", "FLOAT")
|
|
_add_column("signals", "recommended_action", "VARCHAR(50)")
|
|
_add_column("signals", "status", "VARCHAR(50) NOT NULL DEFAULT '待处理'")
|
|
|
|
# qa_records
|
|
_add_column("qa_records", "session_id", "INTEGER")
|
|
_add_column("qa_records", "turn_index", "INTEGER NOT NULL DEFAULT 1")
|
|
|
|
# ai_processing_logs
|
|
_add_column("ai_processing_logs", "normalized_text", "TEXT")
|
|
_add_column("ai_processing_logs", "route", "VARCHAR(80)")
|
|
_add_column("ai_processing_logs", "intent", "VARCHAR(40)")
|
|
_add_column("ai_processing_logs", "executor", "VARCHAR(40)")
|
|
_add_column("ai_processing_logs", "operation_code", "VARCHAR(80)")
|
|
_add_column("ai_processing_logs", "operation_result", "JSON")
|
|
_add_column("ai_processing_logs", "reply_source", "VARCHAR(40)")
|
|
|
|
|
|
def _normalize_dialect(value: str) -> str:
|
|
raw = str(value or "").strip().lower()
|
|
if not raw:
|
|
return ""
|
|
if raw.startswith("postgres"):
|
|
return "postgresql"
|
|
if raw.startswith("sqlite"):
|
|
return "sqlite"
|
|
if raw.startswith("mysql"):
|
|
return "mysql"
|
|
if "+" in raw:
|
|
raw = raw.split("+", 1)[0]
|
|
if ":" in raw:
|
|
raw = raw.split(":", 1)[0]
|
|
return raw
|
|
|
|
|
|
def _mask_db_url(url: str) -> str:
|
|
value = str(url or "").strip()
|
|
if "@" not in value:
|
|
return value
|
|
prefix, suffix = value.split("@", 1)
|
|
if "://" not in prefix:
|
|
return f"***@{suffix}"
|
|
scheme, creds = prefix.split("://", 1)
|
|
user = creds.split(":", 1)[0] if ":" in creds else creds
|
|
return f"{scheme}://{user}:***@{suffix}"
|
|
|
|
|
|
def _check_database_dialect_guard(sync_conn) -> None:
|
|
configured = _normalize_dialect(settings.database_url)
|
|
actual = _normalize_dialect(sync_conn.dialect.name)
|
|
expected = _normalize_dialect(os.getenv("PMFLOW_EXPECTED_DB_DIALECT", ""))
|
|
|
|
env_urls = [
|
|
os.getenv("DATABASE_URL", ""),
|
|
os.getenv("SQLALCHEMY_DATABASE_URL", ""),
|
|
os.getenv("SYNC_DATABASE_URL", ""),
|
|
]
|
|
configured_families = {_normalize_dialect(item) for item in env_urls if str(item or "").strip()}
|
|
configured_families.discard("")
|
|
|
|
logger.info(
|
|
"database_dialect_guard",
|
|
configured=configured,
|
|
actual=actual,
|
|
expected=expected or "unset",
|
|
database_url=_mask_db_url(settings.database_url),
|
|
)
|
|
print(
|
|
"[DB-Guard] "
|
|
f"configured={configured}, actual={actual}, expected={expected or 'unset'}, "
|
|
f"url={_mask_db_url(settings.database_url)}"
|
|
)
|
|
if configured and actual and configured != actual:
|
|
raise RuntimeError(
|
|
f"数据库方言不一致: configured={configured}, actual={actual}。"
|
|
"请确保同一服务只连接单一数据库方言。"
|
|
)
|
|
if expected and actual and expected != actual:
|
|
raise RuntimeError(
|
|
f"数据库方言与 PMFLOW_EXPECTED_DB_DIALECT 不一致: expected={expected}, actual={actual}。"
|
|
)
|
|
if len(configured_families) > 1:
|
|
raise RuntimeError(
|
|
f"检测到多种数据库方言配置: {sorted(configured_families)}。"
|
|
"禁止同一服务混用 SQLite/Postgres。"
|
|
)
|
|
|
|
|
|
def _ensure_feishu_integrity_guards(sync_conn) -> None:
|
|
dialect = _normalize_dialect(sync_conn.dialect.name)
|
|
if dialect not in {"postgresql", "sqlite"}:
|
|
return
|
|
|
|
# Outbox 表兜底，兼容未跑 Alembic 的环境。
|
|
outbox_meta = sa.MetaData()
|
|
outbox_table = sa.Table(
|
|
"feishu_reply_outbox",
|
|
outbox_meta,
|
|
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
|
|
sa.Column("tenant_id", sa.BigInteger(), nullable=False, server_default=sa.text("1")),
|
|
sa.Column("created_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now()),
|
|
sa.Column("updated_at", sa.TIMESTAMP(), nullable=False, server_default=sa.func.now(), onupdate=sa.func.now()),
|
|
sa.Column("trace_id", sa.String(length=64), nullable=True),
|
|
sa.Column("chat_id", sa.String(length=128), nullable=False),
|
|
sa.Column("message_text", sa.Text(), nullable=False),
|
|
sa.Column("status", sa.String(length=20), nullable=False, server_default=sa.text("'pending'")),
|
|
sa.Column("attempts", sa.Integer(), nullable=False, server_default=sa.text("0")),
|
|
sa.Column("next_retry_at", sa.TIMESTAMP(), nullable=True),
|
|
sa.Column("sent_at", sa.TIMESTAMP(), nullable=True),
|
|
sa.Column("last_error", sa.Text(), nullable=True),
|
|
)
|
|
outbox_table.create(sync_conn, checkfirst=True)
|
|
sync_conn.exec_driver_sql(
|
|
"CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_tenant_id ON feishu_reply_outbox (tenant_id)"
|
|
)
|
|
sync_conn.exec_driver_sql(
|
|
"CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_trace_id ON feishu_reply_outbox (trace_id)"
|
|
)
|
|
sync_conn.exec_driver_sql(
|
|
"CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_chat_id ON feishu_reply_outbox (chat_id)"
|
|
)
|
|
sync_conn.exec_driver_sql(
|
|
"CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_status ON feishu_reply_outbox (status)"
|
|
)
|
|
sync_conn.exec_driver_sql(
|
|
"CREATE INDEX IF NOT EXISTS ix_feishu_reply_outbox_next_retry_at ON feishu_reply_outbox (next_retry_at)"
|
|
)
|
|
|
|
inspector = sa.inspect(sync_conn)
|
|
if not inspector.has_table("feishu_items"):
|
|
return
|
|
|
|
duplicate_groups = sync_conn.execute(
|
|
sa.text(
|
|
"""
|
|
SELECT tenant_id, feishu_message_id, split_from_item_id, MIN(id) AS keep_id
|
|
FROM feishu_items
|
|
WHERE feishu_message_id IS NOT NULL
|
|
GROUP BY tenant_id, feishu_message_id, split_from_item_id
|
|
HAVING COUNT(*) > 1
|
|
"""
|
|
)
|
|
).fetchall()
|
|
|
|
table_exists = {
|
|
"signals": inspector.has_table("signals"),
|
|
"project_suggestions": inspector.has_table("project_suggestions"),
|
|
"ai_processing_logs": inspector.has_table("ai_processing_logs"),
|
|
"feishu_items": True,
|
|
}
|
|
|
|
for tenant_id, message_id, split_from_item_id, keep_id in duplicate_groups:
|
|
if split_from_item_id is None:
|
|
duplicate_ids = sync_conn.execute(
|
|
sa.text(
|
|
"""
|
|
SELECT id
|
|
FROM feishu_items
|
|
WHERE tenant_id = :tenant_id
|
|
AND feishu_message_id = :message_id
|
|
AND split_from_item_id IS NULL
|
|
AND id <> :keep_id
|
|
ORDER BY id ASC
|
|
"""
|
|
),
|
|
{
|
|
"tenant_id": int(tenant_id),
|
|
"message_id": str(message_id),
|
|
"keep_id": int(keep_id),
|
|
},
|
|
).fetchall()
|
|
else:
|
|
duplicate_ids = sync_conn.execute(
|
|
sa.text(
|
|
"""
|
|
SELECT id
|
|
FROM feishu_items
|
|
WHERE tenant_id = :tenant_id
|
|
AND feishu_message_id = :message_id
|
|
AND split_from_item_id = :split_from_item_id
|
|
AND id <> :keep_id
|
|
ORDER BY id ASC
|
|
"""
|
|
),
|
|
{
|
|
"tenant_id": int(tenant_id),
|
|
"message_id": str(message_id),
|
|
"split_from_item_id": int(split_from_item_id),
|
|
"keep_id": int(keep_id),
|
|
},
|
|
).fetchall()
|
|
|
|
for (duplicate_id,) in duplicate_ids:
|
|
duplicate_id = int(duplicate_id)
|
|
canonical_id = int(keep_id)
|
|
if table_exists["feishu_items"]:
|
|
sync_conn.execute(
|
|
sa.text(
|
|
"UPDATE feishu_items SET split_from_item_id = :new_id "
|
|
"WHERE split_from_item_id = :old_id"
|
|
),
|
|
{"new_id": canonical_id, "old_id": duplicate_id},
|
|
)
|
|
if table_exists["signals"]:
|
|
sync_conn.execute(
|
|
sa.text(
|
|
"UPDATE signals SET feishu_item_id = :new_id "
|
|
"WHERE feishu_item_id = :old_id"
|
|
),
|
|
{"new_id": canonical_id, "old_id": duplicate_id},
|
|
)
|
|
if table_exists["project_suggestions"]:
|
|
sync_conn.execute(
|
|
sa.text(
|
|
"UPDATE project_suggestions SET feishu_item_id = :new_id "
|
|
"WHERE feishu_item_id = :old_id"
|
|
),
|
|
{"new_id": canonical_id, "old_id": duplicate_id},
|
|
)
|
|
if table_exists["ai_processing_logs"]:
|
|
sync_conn.execute(
|
|
sa.text(
|
|
"UPDATE ai_processing_logs SET feishu_item_id = :new_id "
|
|
"WHERE feishu_item_id = :old_id"
|
|
),
|
|
{"new_id": canonical_id, "old_id": duplicate_id},
|
|
)
|
|
sync_conn.execute(
|
|
sa.text("DELETE FROM feishu_items WHERE id = :id"),
|
|
{"id": duplicate_id},
|
|
)
|
|
|
|
# 幂等唯一约束（含 split 维度）硬化；若存在重复历史数据将直接失败，避免静默错乱。
|
|
sync_conn.exec_driver_sql(
|
|
"""
|
|
CREATE UNIQUE INDEX IF NOT EXISTS uq_feishu_items_tenant_message_root
|
|
ON feishu_items (tenant_id, feishu_message_id)
|
|
WHERE split_from_item_id IS NULL AND feishu_message_id IS NOT NULL
|
|
"""
|
|
)
|
|
sync_conn.exec_driver_sql(
|
|
"""
|
|
CREATE UNIQUE INDEX IF NOT EXISTS uq_feishu_items_tenant_message_split
|
|
ON feishu_items (tenant_id, feishu_message_id, split_from_item_id)
|
|
WHERE split_from_item_id IS NOT NULL AND feishu_message_id IS NOT NULL
|
|
"""
|
|
)
|
|
|
|
|
|
async def _feishu_outbox_drain_loop() -> None:
|
|
while True:
|
|
try:
|
|
dispatched = await dispatch_pending_feishu_replies(limit=50)
|
|
await asyncio.sleep(0.3 if dispatched > 0 else 2.0)
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as exc:
|
|
logger.warning("feishu_outbox_drain_failed", error=str(exc))
|
|
await asyncio.sleep(2.0)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
"""应用生命周期：初始化表与系统技能标签。"""
|
|
auto_create_tables = os.getenv("AUTO_CREATE_TABLES", "true").lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
}
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(_check_database_dialect_guard)
|
|
if auto_create_tables:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
await conn.run_sync(_sqlite_backfill_columns)
|
|
await conn.run_sync(_ensure_feishu_integrity_guards)
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
try:
|
|
created = await init_system_skill_tags(db, tenant_id=1)
|
|
if created:
|
|
logger.info("system_skill_tags_seeded", created=created)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning("seed_system_skills_failed", error=str(exc))
|
|
try:
|
|
interrupted = await AIOrchestrationRunService(db).mark_stale_runs_interrupted(older_than_minutes=10)
|
|
if interrupted:
|
|
await db.commit()
|
|
logger.info("stale_ai_orchestration_runs_interrupted", count=interrupted)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning("interrupt_stale_ai_orchestration_runs_failed", error=str(exc))
|
|
|
|
scheduler = WorkloadRecalcScheduler(AsyncSessionLocal)
|
|
scheduler.start()
|
|
knowledge_scheduler = KnowledgeUrlRefreshScheduler(AsyncSessionLocal)
|
|
knowledge_scheduler.start()
|
|
kg_scheduler = KnowledgeGraphIncrementalScheduler(AsyncSessionLocal)
|
|
kg_scheduler.start()
|
|
ai_pmo_cadence_scheduler = AIPMOCadenceScheduler(AsyncSessionLocal)
|
|
ai_pmo_cadence_scheduler.start()
|
|
feishu_lc = init_feishu_long_connection_manager(AsyncSessionLocal)
|
|
await feishu_lc.start()
|
|
outbox_task = asyncio.create_task(
|
|
_feishu_outbox_drain_loop(),
|
|
name="feishu-outbox-drain-loop",
|
|
)
|
|
|
|
yield
|
|
outbox_task.cancel()
|
|
try:
|
|
await outbox_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
await feishu_lc.stop()
|
|
await ai_pmo_cadence_scheduler.stop()
|
|
await kg_scheduler.stop()
|
|
await knowledge_scheduler.stop()
|
|
await scheduler.stop()
|
|
await engine.dispose()
|
|
|
|
|
|
app = FastAPI(
|
|
title=settings.app_name,
|
|
debug=settings.debug,
|
|
version="0.1.0",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
origins = [item.strip() for item in settings.cors_origins.split(",") if item.strip()]
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins or ["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
app.include_router(api_router)
|
|
|
|
|
|
@app.get("/healthz")
|
|
async def healthz() -> dict:
|
|
return success_response({"status": "ok"})
|