"""数据源的对象索引、探查范围、后台探查任务（docs/design/next-round/03-数据接入与目录维护.md A1）。 由 app/db/models.py 末尾整模块导入，create_all 建表。 **为什么要对象索引。** 以前探查结构每个源最多取 200 个对象、按数据库列出的顺序截断，多出来的表哪儿都看不到，目录、look、 剖析还会把落在窗口外的表说成「表已不在」。现在全库每个对象一行（source_objects，一种方言一条字典查询拿名字、类型、注释， 上限 5 万个），探查范围（数据源 options.introspect_scope 的规则，缺省 500 张、最多 2,000 张）圈出要完整反射的那一部分， 只有范围内的行带着反射到的完整结构（meta）。schema_cache 由 data/source_objects.rebuild_schema_cache 从范围内已反射的行 物化出来，形状不变：消费它的两百多处不用改。 **结构版本。** 每张表一个指纹（反射到的结构的规范 JSON 的哈希）和 struct_version：指纹变了才加 1。第 3 波的结构差异按它 钉版本、出前后差异；这一波只把版本写好，不出变化记录。 **探查任务。** 探查改成后台任务（data/introspect_jobs.py），生命周期是 core/job_runtime 那一套：和同源的起草、剖析排同一条 队（SourceLanes 的 introspect），逐表记状态，能停止，服务重启标中断、不续跑。表照剖析任务的写法：任务一行、逐表一行。 """ from __future__ import annotations from datetime import datetime from typing import Any from sqlalchemy import Boolean, ForeignKey, Index, Integer, String, Text, UniqueConstraint, text from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base, current_owner, PortableJSON, UTCDateTime, new_id, utcnow class SourceObject(Base): """数据源里的一个对象（表或视图）。同一个源同一个 schema 下按名字一行（一个数据源一个 schema，目录-D2）：换了 schema 的源，旧 schema 的行留着不动、不算删除，换回来接着用。 - name：schema_cache["tables"] 的键（SQLAlchemy 规整过的写法：Oracle 全大写的名字是小写）；name_key 按 names.name_key， 范围规则、按名字找表都按它比； - in_scope / scope_reason：在不在探查范围里、为什么（rule 规则命中、explicit 显式列出、pinned 有数据目录或被已发布 模板写死的 SQL 用到、on_demand 助手或运行时按需加入）； - meta：反射到的完整结构（schema_cache 里一张表的那份），只有范围内的才有；离开范围时清掉，指纹和版本留着； - reflect_error：在范围里、却反射不出来（读不到列、权限不够）的原因；这时 meta 是上一次反射成功的那份（有的话）； - removed_at：对象索引里已经没有它了（真删除）。只有它说「表已不在」：范围外、反射失败都不是删除； - added_*：按需加入、在目录页「加入范围」的留痕：谁、经哪个入口（look / db_schema / catalog）、哪一轮对话或哪次运行、 需求原文、什么时候。""" __tablename__ = "source_objects" __table_args__ = ( UniqueConstraint("source_id", "schema_name", "name"), Index("ix_source_objects_scope", "source_id", "schema_name", "in_scope"), Index("ix_source_objects_run", "added_run_id"), ) id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) #: 这一行是哪个 schema 的（"" 是连接的默认 schema） schema_name: Mapped[str] = mapped_column(String(255), default="") name: Mapped[str] = mapped_column(String(255)) name_key: Mapped[str] = mapped_column(String(255)) #: table | view kind: Mapped[str] = mapped_column(String(10), default="table") comment: Mapped[str | None] = mapped_column(Text, default=None) #: Oracle 库里本来就是小写的名字（SQLAlchemy 反射回来是 quoted_name(quote=True)）：存成字符串就丢了这个标记，反射时 #: 要按它原样加引号，不然 Oracle 转成大写去找，找的是另一个对象 quote_name: Mapped[bool] = mapped_column(Boolean, default=False) #: 数据字典里的行数估算（MySQL TABLE_ROWS、PostgreSQL reltuples、Oracle NUM_ROWS）；不知道是 NULL est_rows: Mapped[int | None] = mapped_column(Integer, default=None) in_scope: Mapped[bool] = mapped_column(Boolean, default=False) scope_reason: Mapped[str | None] = mapped_column(String(20), default=None) meta: Mapped[dict[str, Any] | None] = mapped_column(default=None) fingerprint: Mapped[str | None] = mapped_column(String(64), default=None) struct_version: Mapped[int] = mapped_column(Integer, default=0) reflect_error: Mapped[str | None] = mapped_column(Text, default=None) first_seen_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) last_seen_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) removed_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) reflected_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) #: look | db_schema | catalog added_via: Mapped[str | None] = mapped_column(String(20), default=None) added_by: Mapped[str | None] = mapped_column(String(100), default=None) added_run_id: Mapped[str | None] = mapped_column(String(32), default=None) added_conversation_id: Mapped[str | None] = mapped_column(String(32), default=None) #: look 记录的编号（L1…）、运行里的节点 id：同一轮、同一次运行里是哪一下加的 added_ref: Mapped[str | None] = mapped_column(String(64), default=None) added_need: Mapped[str | None] = mapped_column(Text, default=None) added_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) class IntrospectJob(Base): """一次后台探查。建任务的请求里已经刷新了对象索引、按范围规则圈好了表、核过上限（超了 409，不建任务）；任务只做反射： 逐表（按批）读结构、写进对象索引的行，最后由 rebuild_schema_cache 物化 schema_cache。 同一个源同时只有一个排队或进行中的探查任务（接口先查一遍，部分唯一索引兜底）；和同源的起草、剖析按建立的先后排队。 挂外键、级联删除：任务是数据源的附属记录。""" __tablename__ = "introspect_jobs" __table_args__ = ( Index("ix_introspect_jobs_source_created", "source_id", "created_at"), Index("ux_introspect_jobs_active", "source_id", unique=True, sqlite_where=text("status IN ('queued', 'running')"), postgresql_where=text("status IN ('queued', 'running')")), ) id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) # queued | running | succeeded | partial | failed | cancelled | interrupted status: Mapped[str] = mapped_column(String(20), default="queued") #: 认领这一行的工作进程（app/cluster；缺省是建它的进程）。多进程时启动整理、接手只碰属主不在了的行 owner: Mapped[str | None] = mapped_column(String(120), default=current_owner) #: manual 人点的 | schedule 定时维护（第 3 波） trigger: Mapped[str] = mapped_column(String(20), default="manual") #: 这次探查的是哪个 schema（"" 是默认 schema） schema_name: Mapped[str] = mapped_column(String(255), default="") #: 建任务那一刻的范围：{rules（规则原样）, in_scope, rule, pinned, explicit, on_demand, out_of_scope, cap, index_total, #: index_method, index_capped, removed, added}——留痕：这一次按什么规则、圈了多少张 scope: Mapped[dict[str, Any]] = mapped_column(default=dict) created_by: Mapped[str | None] = mapped_column(String(100), default=None) created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) #: 整个任务出错、数据源中途停用时的整句原因 error: Mapped[str | None] = mapped_column(Text, default=None) class IntrospectTask(Base): """探查任务里的一张表。按 seq 排。状态：pending 排队、running 正在读、ok 读到了、failed 读不到（原因在 error）、 cancelled 停止或服务重启时还没读。""" __tablename__ = "introspect_tasks" __table_args__ = (UniqueConstraint("job_id", "seq"),) id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) job_id: Mapped[str] = mapped_column(ForeignKey("introspect_jobs.id", ondelete="CASCADE"), index=True) seq: Mapped[int] = mapped_column(Integer) table_name: Mapped[str] = mapped_column(String(255)) #: 为什么在范围里（pinned / explicit / rule / on_demand）：开始执行时按它把范围写进对象索引 reason: Mapped[str | None] = mapped_column(String(20), default=None) status: Mapped[str] = mapped_column(String(20), default="pending") error: Mapped[str | None] = mapped_column(Text, default=None) #: 结构版本：读完以后的（指纹没变就和以前一样） struct_version: Mapped[int] = mapped_column(Integer, default=0) finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) # -------------------------------------------------------------------------- # 第 3 波：取值观察、变化记录（docs/design/next-round/03-数据接入与目录维护.md B2、B3） # -------------------------------------------------------------------------- class TableObservation(Base): """一次取值观察（剖析引擎的四类观察检查，data/catalog_profile_observe.py）：码值列的取值分布（codes_observe）、最新业务 日期（freshness）、最近 35 天按天的行数（daily_volume）、行数（rows）。 **为什么单独一张表。** 剖析任务的逐表一行只留最近 50 个任务，而且一张表一次剖析只有一份发现；取值漂移要拿同一口径的 历史比（新码值、比例剧变、断档、缺天、量跌、行数跌），得按「表 × 种类 × 列」留一串。每张表每种观察（码值按列）留最近 KEEP 条，更早的删掉。 **口径（basis）。** 每条都记它看的是哪一部分数据：full 整表、first_rows（按存储顺序的前 N 行）、window（业务日期最近 N 天）、 sample（块采样 p%）、stats / count（行数的来源）。basis_key 是口径的规范写法：只和 basis_key 相同的历史比，口径不同的照实说 不比。遮罩列不出现在任何一条观察查询里，取值也就不会进这张表。 挂外键、级联删除：观察是数据源的附属记录。""" __tablename__ = "table_observations" __table_args__ = ( Index("ix_table_observations_lookup", "source_id", "table_name", "kind", "column_name", "at"), ) #: 每张表每种观察（码值按列）留几条 KEEP = 26 id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) table_name: Mapped[str] = mapped_column(String(255)) #: 码值、业务日期观察的那一列；表级的（行数）是空串 column_name: Mapped[str] = mapped_column(String(255), default="") #: codes_observe | freshness | daily_volume | rows kind: Mapped[str] = mapped_column(String(20)) basis: Mapped[dict[str, Any]] = mapped_column(default=dict) basis_key: Mapped[str] = mapped_column(String(200), default="") #: 观察到的：码值 {values: [[码, 行数]…], rows, truncated}；最新业务日期 {max}；按日 {since, until, days: {日: 行数}}； #: 行数 {rows, at_least, method} data: Mapped[dict[str, Any]] = mapped_column(default=dict) #: 哪个剖析任务的第几张表测到的 job_id: Mapped[str | None] = mapped_column(String(32), default=None) task_seq: Mapped[int | None] = mapped_column(Integer, default=None) at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) class SourceChange(Base): """数据源的一条变化（data/source_changes.py，格式 agentlab-change/1）：结构变化（重新探查后的前后差异）和取值变化（取值 观察和同口径历史的比较）。全站只有这一份记录：审阅页的「变化」、数据源卡片的角标、外部通知、口径账本的召回都读它。 - seq：自增，不复用（sqlite_autoincrement），召回按它当游标拉取；id 是给界面、链接用的编号； - kind：结构类 table_added / table_removed / table_left_scope / table_reflect_failed / column_added / column_removed / column_type_changed / column_nullable_changed / key_changed / fk_changed / comment_changed / scope_changed；取值类 codes_new / codes_shift / date_stalled / date_gap / volume_drop / rows_drop / rows_jump / coverage_drop； - severity：info / warning / action（现在大概率已经不对了：已确认的项、写死的 SQL 指向被删的列，或者「已列全」被撤销）； - detected_by：{job_kind, job_id, task_seq, trigger: manual|schedule|on_demand, actor}；basis：{method, query_refs}； - before / after：前后各是什么（码值只记非遮罩列的）；affects：{catalog: [{table, path, status, source}], templates: [{workflow_id, name, version, impact, nodes}], rules_touched}；struct_version_before / after：结构变化前后的结构版本； - fingerprint：去重——同一个变化再次测到只更新 last_seen_at 和 seen_count； - status：open / acknowledged / resolved / dismissed；resolution：{by, at, action, note, catalog_version, job_id}。 挂外键、级联删除：变化是数据源的附属记录。""" __tablename__ = "source_changes" __table_args__ = ( UniqueConstraint("source_id", "fingerprint"), Index("ix_source_changes_source_status", "source_id", "status"), Index("ix_source_changes_source_table", "source_id", "table_name"), {"sqlite_autoincrement": True}, ) seq: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[str] = mapped_column(String(32), unique=True, default=new_id) source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) #: 测到时这个源叫什么（之后改了名，这条记录说的仍是当时的名字） source_name: Mapped[str] = mapped_column(String(100), default="") #: 源级的变化（范围规则改了）是空串 table_name: Mapped[str] = mapped_column(String(255), default="") column_name: Mapped[str | None] = mapped_column(String(255), default=None) kind: Mapped[str] = mapped_column(String(40)) severity: Mapped[str] = mapped_column(String(10), default="info") detected_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) last_seen_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) seen_count: Mapped[int] = mapped_column(Integer, default=1) detected_by: Mapped[dict[str, Any]] = mapped_column(default=dict) basis: Mapped[dict[str, Any] | None] = mapped_column(default=None) before: Mapped[Any] = mapped_column(PortableJSON, nullable=True, default=None) after: Mapped[Any] = mapped_column(PortableJSON, nullable=True, default=None) affects: Mapped[dict[str, Any] | None] = mapped_column(default=None) struct_version_before: Mapped[int | None] = mapped_column(Integer, default=None) struct_version_after: Mapped[int | None] = mapped_column(Integer, default=None) fingerprint: Mapped[str] = mapped_column(String(64)) status: Mapped[str] = mapped_column(String(20), default="open") resolution: Mapped[dict[str, Any] | None] = mapped_column(default=None) #: 记录的格式（agentlab-change/1）：召回按它读，字段以后改要升版本 schema_tag: Mapped[str] = mapped_column("schema", String(30), default="agentlab-change/1") __all__ = ["IntrospectJob", "IntrospectTask", "SourceChange", "SourceObject", "TableObservation"]