from __future__ import annotations from typing import Any from sqlalchemy.dialects.postgresql import JSONB, UUID from app import db from app.core.common.identifiers import new_governance_uid from app.core.common.timezone_utils import now_china, now_china_naive JOB_STATUSES = ( "created", "queued", "extracting", "normalizing", "matching", "awaiting_review", "published", "partial", "failed", "cancelled", ) def _iso(value: Any) -> str | None: return value.isoformat() if value is not None else None class IngestionSource(db.Model): __tablename__ = "ingestion_sources" __table_args__ = ( db.CheckConstraint( "source_type IN ('database','file','ddl')", name="ck_ingestion_source_type", ), db.CheckConstraint( "status IN ('active','disabled')", name="ck_ingestion_source_status", ), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) source_type = db.Column(db.String(20), nullable=False) name = db.Column(db.String(300), nullable=False) config = db.Column(JSONB, nullable=False, default=dict) permission_scope = db.Column(JSONB, nullable=False, default=dict) status = db.Column(db.String(20), nullable=False, default="active") created_by = db.Column(db.String(100)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) def to_dict(self) -> dict[str, Any]: return { "uid": str(self.uid), "source_type": self.source_type, "name": self.name, "permission_scope": dict(self.permission_scope or {}), "status": self.status, "created_by": self.created_by, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), } class SourceArtifact(db.Model): __tablename__ = "source_artifacts" __table_args__ = ( db.CheckConstraint("size_bytes >= 0", name="ck_source_artifact_size"), db.UniqueConstraint( "source_uid", "content_hash", "parser_version", name="uq_source_artifact_parser_content", ), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) source_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_sources.uid"), nullable=False, ) filename = db.Column(db.String(500), nullable=False) media_type = db.Column(db.String(200), nullable=False) size_bytes = db.Column(db.BigInteger, nullable=False) content_hash = db.Column(db.String(64), nullable=False) storage_ref = db.Column(db.Text, nullable=False) parser_version = db.Column(db.String(100), nullable=False) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) def to_dict(self) -> dict[str, Any]: return { "uid": str(self.uid), "source_uid": str(self.source_uid), "filename": self.filename, "media_type": self.media_type, "size_bytes": int(self.size_bytes), "content_hash": self.content_hash, "parser_version": self.parser_version, "created_at": _iso(self.created_at), } class IngestionJob(db.Model): __tablename__ = "ingestion_jobs" __table_args__ = ( db.CheckConstraint( "status IN (" + ",".join(f"'{value}'" for value in JOB_STATUSES) + ")", name="ck_ingestion_job_status", ), db.UniqueConstraint( "idempotency_key", name="uq_ingestion_job_idempotency_key", ), db.CheckConstraint( "attempt_count >= 0", name="ck_ingestion_job_attempt_count", ), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) source_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_sources.uid"), nullable=False, ) artifact_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.source_artifacts.uid"), ) job_type = db.Column(db.String(30), nullable=False) status = db.Column(db.String(30), nullable=False, default="created") idempotency_key = db.Column(db.String(64), nullable=False, unique=True) parser_version = db.Column(db.String(100), nullable=False) parameters = db.Column(JSONB, nullable=False, default=dict) statistics = db.Column(JSONB, nullable=False, default=dict) last_error = db.Column(db.String(1000)) attempt_count = db.Column(db.Integer, nullable=False, default=0) failure_stage = db.Column(db.String(30)) actor_uid = db.Column(db.String(100)) force_rerun = db.Column(db.Boolean, nullable=False, default=False) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) started_at = db.Column(db.DateTime) finished_at = db.Column(db.DateTime) def to_dict(self) -> dict[str, Any]: return { "uid": str(self.uid), "source_uid": str(self.source_uid), "artifact_uid": str(self.artifact_uid) if self.artifact_uid else None, "job_type": self.job_type, "status": self.status, "parser_version": self.parser_version, "parameters": dict(self.parameters or {}), "statistics": dict(self.statistics or {}), "last_error": self.last_error, "attempt_count": int(self.attempt_count or 0), "failure_stage": self.failure_stage, "actor_uid": self.actor_uid, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), "started_at": _iso(self.started_at), "finished_at": _iso(self.finished_at), } class CatalogSnapshot(db.Model): __tablename__ = "catalog_snapshots" __table_args__ = ( db.CheckConstraint( "attempt > 0", name="ck_catalog_snapshot_attempt", ), db.UniqueConstraint( "job_uid", "attempt", name="uq_catalog_snapshot_job_attempt", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) job_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_jobs.uid", ondelete="CASCADE"), nullable=False, ) source_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_sources.uid"), nullable=False, ) attempt = db.Column(db.Integer, nullable=False) database_type = db.Column(db.String(20), nullable=False) content_hash = db.Column(db.String(64), nullable=False) snapshot = db.Column(JSONB, nullable=False) created_at = db.Column( db.DateTime, nullable=False, default=now_china_naive, ) def to_dict(self) -> dict[str, Any]: return { "uid": str(self.uid), "job_uid": str(self.job_uid), "source_uid": str(self.source_uid), "attempt": int(self.attempt), "database_type": self.database_type, "content_hash": self.content_hash, "snapshot": dict(self.snapshot or {}), "created_at": _iso(self.created_at), } class DeviceAsset(db.Model): __tablename__ = "device_assets" __table_args__ = ( db.CheckConstraint( "asset_type IN (" "'device','component','measurement_point'," "'alarm','maintenance_record'" ")", name="ck_device_asset_type", ), db.CheckConstraint( "status IN ('active','retired')", name="ck_device_asset_status", ), db.CheckConstraint( "current_version > 0", name="ck_device_asset_current_version", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) asset_type = db.Column(db.String(40), nullable=False) name = db.Column(db.String(300), nullable=False) status = db.Column(db.String(20), nullable=False, default="active") current_version = db.Column(db.Integer, nullable=False, default=1) content_hash = db.Column(db.String(64), nullable=False) location = db.Column(db.String(300)) organization = db.Column(db.String(300)) responsible_person = db.Column(db.String(300)) attributes = db.Column(JSONB, nullable=False, default=dict) created_by = db.Column(db.String(100)) updated_by = db.Column(db.String(100)) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) updated_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceAssetSourceMapping(db.Model): __tablename__ = "device_asset_source_mappings" __table_args__ = ( db.CheckConstraint( "asset_type IN (" "'device','component','measurement_point'," "'alarm','maintenance_record'" ")", name="ck_device_asset_mapping_type", ), db.UniqueConstraint( "source_uid", "source_entity", "asset_type", "source_code", name="uq_device_asset_source_identity", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="CASCADE"), nullable=False, ) source_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_sources.uid"), nullable=False, ) source_entity = db.Column(db.String(300), nullable=False) asset_type = db.Column(db.String(40), nullable=False) source_code = db.Column(db.String(300), nullable=False) source_updated_at = db.Column(db.DateTime(timezone=True)) first_seen_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) last_seen_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceAssetVersion(db.Model): __tablename__ = "device_asset_versions" __table_args__ = ( db.CheckConstraint( "version > 0", name="ck_device_asset_version_number", ), db.UniqueConstraint( "asset_uid", "version", name="uq_device_asset_version", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="CASCADE"), nullable=False, ) version = db.Column(db.Integer, nullable=False) content_hash = db.Column(db.String(64), nullable=False) snapshot = db.Column(JSONB, nullable=False) source_mapping_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_asset_source_mappings.uid", ondelete="RESTRICT", ), nullable=False, ) actor_uid = db.Column(db.String(100)) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceSemanticCode(db.Model): __tablename__ = "device_semantic_codes" __table_args__ = ( db.CheckConstraint( "code_type IN ('fault','cause','action')", name="ck_device_semantic_code_type", ), db.CheckConstraint( "status IN (" "'draft','in_review','published','rejected','retired'" ")", name="ck_device_semantic_code_status", ), db.CheckConstraint( "current_version > 0", name="ck_device_semantic_code_current_version", ), db.CheckConstraint( "suggestion_source IN ('manual','rule','ai')", name="ck_device_semantic_suggestion_source", ), db.CheckConstraint( "confidence IS NULL OR (confidence >= 0 AND confidence <= 1)", name="ck_device_semantic_confidence", ), db.UniqueConstraint( "ontology_uid", "code_type", "canonical_code", name="uq_device_semantic_identity", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) ontology_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid", ondelete="RESTRICT"), nullable=False, ) code_type = db.Column(db.String(20), nullable=False) canonical_code = db.Column(db.String(120), nullable=False) canonical_name = db.Column(db.String(300), nullable=False) definition = db.Column(db.Text) status = db.Column(db.String(20), nullable=False, default="draft") current_version = db.Column(db.Integer, nullable=False, default=1) source_mappings = db.Column(JSONB, nullable=False, default=list) evidence_uids = db.Column(JSONB, nullable=False, default=list) suggestion_source = db.Column( db.String(20), nullable=False, default="manual", ) confidence = db.Column(db.Float) created_by = db.Column(db.String(100)) updated_by = db.Column(db.String(100)) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) updated_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceSemanticCodeVersion(db.Model): __tablename__ = "device_semantic_code_versions" __table_args__ = ( db.CheckConstraint( "version > 0", name="ck_device_semantic_code_version_number", ), db.UniqueConstraint( "code_uid", "version", name="uq_device_semantic_code_version", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) code_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_semantic_codes.uid", ondelete="CASCADE", ), nullable=False, ) version = db.Column(db.Integer, nullable=False) snapshot = db.Column(JSONB, nullable=False) created_by = db.Column(db.String(100)) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceSemanticCodeReview(db.Model): __tablename__ = "device_semantic_code_reviews" __table_args__ = ( db.CheckConstraint( "version > 0", name="ck_device_semantic_review_version", ), db.CheckConstraint( "decision IN ('approve','reject')", name="ck_device_semantic_review_decision", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) code_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_semantic_codes.uid", ondelete="CASCADE", ), nullable=False, ) version = db.Column(db.Integer, nullable=False) decision = db.Column(db.String(20), nullable=False) reason = db.Column(db.String(1000), nullable=False, default="") actor_uid = db.Column(db.String(100), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceEntityMatchCandidate(db.Model): __tablename__ = "device_entity_match_candidates" __table_args__ = ( db.CheckConstraint( "left_asset_uid <> right_asset_uid", name="ck_device_entity_candidate_distinct_assets", ), db.CheckConstraint( "status IN ('pending','merged','rejected','rolled_back')", name="ck_device_entity_candidate_status", ), db.CheckConstraint( "suggestion_source IN ('rule','ai','manual')", name="ck_device_entity_candidate_source", ), db.CheckConstraint( "confidence >= 0 AND confidence <= 1", name="ck_device_entity_candidate_confidence", ), db.CheckConstraint( "current_version > 0", name="ck_device_entity_candidate_version", ), db.Index( "uq_device_entity_open_pair", "left_asset_uid", "right_asset_uid", unique=True, postgresql_where=db.text( "status IN ('pending','merged')" ), ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) left_asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) right_asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) canonical_asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), ) status = db.Column(db.String(20), nullable=False, default="pending") suggestion_source = db.Column(db.String(20), nullable=False) confidence = db.Column(db.Float, nullable=False) explanation = db.Column(JSONB, nullable=False, default=list) evidence_uids = db.Column(JSONB, nullable=False, default=list) model_provider = db.Column(db.String(200)) model_name = db.Column(db.String(200)) current_version = db.Column(db.Integer, nullable=False, default=1) created_by = db.Column(db.String(100)) reviewed_by = db.Column(db.String(100)) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) updated_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceEntityMatchReview(db.Model): __tablename__ = "device_entity_match_reviews" __table_args__ = ( db.CheckConstraint( "version > 0", name="ck_device_entity_review_version", ), db.CheckConstraint( "decision IN ('approve','reject','auto_approve')", name="ck_device_entity_review_decision", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) candidate_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_entity_match_candidates.uid", ondelete="CASCADE", ), nullable=False, ) version = db.Column(db.Integer, nullable=False) decision = db.Column(db.String(20), nullable=False) reason = db.Column(db.String(1000), nullable=False) actor_uid = db.Column(db.String(100), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceEntityMergeEvent(db.Model): __tablename__ = "device_entity_merge_events" __table_args__ = ( db.UniqueConstraint( "candidate_uid", name="uq_device_entity_merge_candidate", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) candidate_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_entity_match_candidates.uid", ondelete="CASCADE", ), nullable=False, ) canonical_asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) member_asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) review_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_entity_match_reviews.uid", ondelete="RESTRICT", ), nullable=False, ) snapshot = db.Column(JSONB, nullable=False) actor_uid = db.Column(db.String(100), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceEntityMergeRollback(db.Model): __tablename__ = "device_entity_merge_rollbacks" __table_args__ = ( db.UniqueConstraint( "merge_uid", name="uq_device_entity_merge_rollback", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) merge_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_entity_merge_events.uid", ondelete="CASCADE", ), nullable=False, ) candidate_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_entity_match_candidates.uid", ondelete="CASCADE", ), nullable=False, ) reason = db.Column(db.String(1000), nullable=False) snapshot = db.Column(JSONB, nullable=False) actor_uid = db.Column(db.String(100), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceQualityProfile(db.Model): __tablename__ = "device_quality_profiles" __table_args__ = ({"schema": "public"},) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) code = db.Column(db.String(120), nullable=False, unique=True) name = db.Column(db.String(300), nullable=False) created_by = db.Column(db.String(120), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceQualityProfileVersion(db.Model): __tablename__ = "device_quality_profile_versions" __table_args__ = ( db.CheckConstraint( "version > 0", name="ck_device_quality_profile_version_number", ), db.CheckConstraint( "status IN ('draft','published','superseded')", name="ck_device_quality_profile_version_status", ), db.UniqueConstraint( "profile_uid", "version", name="uq_device_quality_profile_version", ), db.UniqueConstraint( "profile_uid", "content_hash", name="uq_device_quality_profile_hash", ), db.Index( "uq_device_quality_active_profile", "profile_uid", unique=True, postgresql_where=db.text("status = 'published'"), ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) profile_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_quality_profiles.uid", ondelete="RESTRICT", ), nullable=False, ) version = db.Column(db.Integer, nullable=False) status = db.Column(db.String(20), nullable=False, default="draft") rules = db.Column(JSONB, nullable=False) content_hash = db.Column(db.String(64), nullable=False) created_by = db.Column(db.String(120), nullable=False) published_by = db.Column(db.String(120)) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) published_at = db.Column(db.DateTime(timezone=True)) class DeviceQualityRun(db.Model): __tablename__ = "device_quality_runs" __table_args__ = ( db.CheckConstraint( "status IN ('success','failed')", name="ck_device_quality_run_status", ), db.CheckConstraint( "total_assets >= 0 AND total_violations >= 0", name="ck_device_quality_run_counts", ), db.CheckConstraint( "score >= 0 AND score <= 100", name="ck_device_quality_run_score", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) policy_version_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_quality_profile_versions.uid", ondelete="RESTRICT", ), nullable=False, ) policy_hash = db.Column(db.String(64), nullable=False) source_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.ingestion_sources.uid", ondelete="RESTRICT", ), ) status = db.Column(db.String(20), nullable=False) total_assets = db.Column(db.Integer, nullable=False) total_violations = db.Column(db.Integer, nullable=False) score = db.Column(db.Float, nullable=False) created_by = db.Column(db.String(120), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceQualityRuleResult(db.Model): __tablename__ = "device_quality_rule_results" __table_args__ = ( db.CheckConstraint( "severity IN ('info','warning','error','critical')", name="ck_device_quality_rule_result_severity", ), db.CheckConstraint( "status IN ('passed','violated','not_applicable')", name="ck_device_quality_rule_result_status", ), db.CheckConstraint( "evaluated_count >= 0 AND violation_count >= 0 " "AND sampled_count >= 0 " "AND violation_count <= evaluated_count " "AND sampled_count <= violation_count", name="ck_device_quality_rule_result_counts", ), db.CheckConstraint( "pass_rate >= 0 AND pass_rate <= 1 " "AND weighted_score >= 0 AND weighted_score <= weight " "AND weight >= 0 AND weight <= 100", name="ck_device_quality_rule_result_scores", ), db.UniqueConstraint( "run_uid", "rule_code", name="uq_device_quality_run_rule", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) run_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_quality_runs.uid", ondelete="CASCADE"), nullable=False, ) rule_code = db.Column(db.String(120), nullable=False) severity = db.Column(db.String(20), nullable=False) weight = db.Column(db.Float, nullable=False) status = db.Column(db.String(30), nullable=False) evaluated_count = db.Column(db.Integer, nullable=False) violation_count = db.Column(db.Integer, nullable=False) sampled_count = db.Column(db.Integer, nullable=False) pass_rate = db.Column(db.Float, nullable=False) weighted_score = db.Column(db.Float, nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceQualityViolationSample(db.Model): __tablename__ = "device_quality_violation_samples" __table_args__ = ( db.CheckConstraint( "severity IN ('info','warning','error','critical')", name="ck_device_quality_violation_severity", ), db.UniqueConstraint( "run_uid", "rule_code", "asset_uid", "field_name", name="uq_device_quality_violation_identity", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) run_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_quality_runs.uid", ondelete="CASCADE"), nullable=False, ) rule_code = db.Column(db.String(120), nullable=False) severity = db.Column(db.String(20), nullable=False) asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) field_name = db.Column(db.String(300), nullable=False) source_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.ingestion_sources.uid", ondelete="RESTRICT", ), ) source_mapping_uid = db.Column( UUID(as_uuid=False), db.ForeignKey( "public.device_asset_source_mappings.uid", ondelete="RESTRICT", ), ) message = db.Column(db.String(1000), nullable=False) evidence = db.Column(JSONB, nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) expires_at = db.Column(db.DateTime(timezone=True), nullable=False) class DeviceQualityAssetScore(db.Model): __tablename__ = "device_quality_asset_scores" __table_args__ = ( db.CheckConstraint( "status IN ('passed','violated','not_applicable')", name="ck_device_quality_asset_score_status", ), db.CheckConstraint( "evaluated_rule_count >= 0 AND violation_count >= 0 " "AND violation_count <= evaluated_rule_count", name="ck_device_quality_asset_score_counts", ), db.CheckConstraint( "score >= 0 AND score <= 100", name="ck_device_quality_asset_score_value", ), db.UniqueConstraint( "run_uid", "asset_uid", name="uq_device_quality_run_asset", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) run_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_quality_runs.uid", ondelete="CASCADE"), nullable=False, ) asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) asset_type = db.Column(db.String(40), nullable=False) status = db.Column(db.String(30), nullable=False) evaluated_rule_count = db.Column(db.Integer, nullable=False) violation_count = db.Column(db.Integer, nullable=False) score = db.Column(db.Float, nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceQualityIssue(db.Model): __tablename__ = "device_quality_issues" __table_args__ = ( db.CheckConstraint( "severity IN ('info','warning','error','critical')", name="ck_device_quality_issue_severity", ), db.CheckConstraint( "priority IN ('low','medium','high','critical')", name="ck_device_quality_issue_priority", ), db.CheckConstraint( "status IN " "('open','assigned','in_progress','pending_review','closed')", name="ck_device_quality_issue_status", ), db.CheckConstraint( "occurrence_number > 0 AND current_version > 0", name="ck_device_quality_issue_versions", ), db.Index( "uq_device_quality_open_recurrence", "recurrence_key", unique=True, postgresql_where=db.text("status <> 'closed'"), ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) issue_code = db.Column(db.String(32), nullable=False, unique=True) source_violation_uid = db.Column(UUID(as_uuid=False), nullable=False) source_run_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_quality_runs.uid", ondelete="RESTRICT"), nullable=False, ) rule_code = db.Column(db.String(120), nullable=False) severity = db.Column(db.String(20), nullable=False) priority = db.Column(db.String(20), nullable=False) asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) field_name = db.Column(db.String(300), nullable=False) source_uid = db.Column(UUID(as_uuid=False)) source_mapping_uid = db.Column(UUID(as_uuid=False)) message = db.Column(db.String(1000), nullable=False) evidence = db.Column(JSONB, nullable=False) recurrence_key = db.Column(db.String(64), nullable=False) occurrence_number = db.Column(db.Integer, nullable=False) status = db.Column(db.String(30), nullable=False, default="open") assignee_uid = db.Column(UUID(as_uuid=False)) due_at = db.Column(db.DateTime(timezone=True)) current_version = db.Column(db.Integer, nullable=False, default=1) created_by = db.Column(UUID(as_uuid=False), nullable=False) updated_by = db.Column(UUID(as_uuid=False), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) updated_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) closed_at = db.Column(db.DateTime(timezone=True)) class DeviceQualityIssueRemediation(db.Model): __tablename__ = "device_quality_issue_remediations" __table_args__ = ( db.CheckConstraint( "round_number > 0", name="ck_device_quality_issue_remediation_round", ), db.CheckConstraint( "review_status IN ('pending','passed','failed')", name="ck_device_quality_issue_remediation_review", ), db.UniqueConstraint( "issue_uid", "round_number", name="uq_device_quality_issue_remediation_round", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) issue_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_quality_issues.uid", ondelete="CASCADE"), nullable=False, ) round_number = db.Column(db.Integer, nullable=False) summary = db.Column(db.String(2000), nullable=False) evidence_refs = db.Column(JSONB, nullable=False) submitted_by = db.Column(UUID(as_uuid=False), nullable=False) submitted_at = db.Column(db.DateTime(timezone=True), nullable=False) review_status = db.Column(db.String(20), nullable=False, default="pending") reviewed_by = db.Column(UUID(as_uuid=False)) reviewed_at = db.Column(db.DateTime(timezone=True)) review_note = db.Column(db.String(2000)) verification_run_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_quality_runs.uid", ondelete="RESTRICT"), ) class DeviceQualityIssueTimeline(db.Model): __tablename__ = "device_quality_issue_timeline" __table_args__ = ( db.CheckConstraint( "to_status IN " "('open','assigned','in_progress','pending_review','closed')", name="ck_device_quality_issue_timeline_status", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) issue_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_quality_issues.uid", ondelete="CASCADE"), nullable=False, ) action = db.Column(db.String(40), nullable=False) from_status = db.Column(db.String(30)) to_status = db.Column(db.String(30), nullable=False) actor_uid = db.Column(UUID(as_uuid=False), nullable=False) note = db.Column(db.String(2000)) payload = db.Column(JSONB, nullable=False) created_at = db.Column(db.DateTime(timezone=True), nullable=False) class DeviceOperationalEvent(db.Model): __tablename__ = "device_operational_events" __table_args__ = ( db.CheckConstraint( "event_type IN ('alarm','fault','maintenance','downtime')", name="ck_device_operational_event_type", ), db.CheckConstraint( "severity IN ('info','warning','error','critical')", name="ck_device_operational_event_severity", ), db.CheckConstraint( "status IN ('observed','resolved')", name="ck_device_operational_event_status", ), db.CheckConstraint( "ended_at IS NULL OR ended_at >= occurred_at", name="ck_device_operational_event_window", ), db.UniqueConstraint( "source_uid", "source_entity", "source_code", name="uq_device_operational_event_source_identity", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) source_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_sources.uid", ondelete="RESTRICT"), nullable=False, ) source_entity = db.Column(db.String(300), nullable=False) source_code = db.Column(db.String(300), nullable=False) event_type = db.Column(db.String(30), nullable=False) asset_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), nullable=False, ) component_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.device_assets.uid", ondelete="RESTRICT"), ) title = db.Column(db.String(500), nullable=False) severity = db.Column(db.String(20), nullable=False) status = db.Column(db.String(20), nullable=False) occurred_at = db.Column(db.DateTime(timezone=True), nullable=False) ended_at = db.Column(db.DateTime(timezone=True)) evidence_refs = db.Column(JSONB, nullable=False) content_hash = db.Column(db.String(64), nullable=False) created_by = db.Column(UUID(as_uuid=False), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class DeviceEvidenceRelation(db.Model): __tablename__ = "device_evidence_relations" __table_args__ = ( db.CheckConstraint( "from_kind IN ('asset','event','quality_issue')", name="ck_device_evidence_relation_from_kind", ), db.CheckConstraint( "to_kind IN ('asset','event','quality_issue')", name="ck_device_evidence_relation_to_kind", ), db.CheckConstraint( "relation_type IN (" "'part_of','occurred_on','indicates','triggered'," "'addresses','impacts','evidences','related_to'" ")", name="ck_device_evidence_relation_type", ), db.CheckConstraint( "source IN ('imported','manual','derived')", name="ck_device_evidence_relation_source", ), db.CheckConstraint( "NOT (from_kind = to_kind AND from_uid = to_uid)", name="ck_device_evidence_relation_distinct_nodes", ), db.UniqueConstraint( "from_kind", "from_uid", "relation_type", "to_kind", "to_uid", name="uq_device_evidence_relation_identity", ), {"schema": "public"}, ) uid = db.Column( UUID(as_uuid=False), primary_key=True, default=new_governance_uid, ) from_kind = db.Column(db.String(30), nullable=False) from_uid = db.Column(UUID(as_uuid=False), nullable=False) relation_type = db.Column(db.String(40), nullable=False) to_kind = db.Column(db.String(30), nullable=False) to_uid = db.Column(UUID(as_uuid=False), nullable=False) evidence_refs = db.Column(JSONB, nullable=False) source = db.Column(db.String(30), nullable=False) created_by = db.Column(UUID(as_uuid=False), nullable=False) created_at = db.Column( db.DateTime(timezone=True), nullable=False, default=now_china, ) class EvidenceFragment(db.Model): __tablename__ = "evidence_fragments" __table_args__ = ( db.CheckConstraint( "confidence IS NULL OR (confidence >= 0 AND confidence <= 1)", name="ck_evidence_confidence", ), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) job_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_jobs.uid", ondelete="CASCADE"), nullable=False, ) artifact_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.source_artifacts.uid"), ) locator = db.Column(JSONB, nullable=False) excerpt = db.Column(db.Text) confidence = db.Column(db.Float) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) def to_dict(self) -> dict[str, Any]: return { "uid": str(self.uid), "job_uid": str(self.job_uid), "artifact_uid": str(self.artifact_uid) if self.artifact_uid else None, "locator": dict(self.locator or {}), "excerpt": self.excerpt, "confidence": self.confidence, "created_at": _iso(self.created_at), } class ExtractionCandidate(db.Model): __tablename__ = "extraction_candidates" __table_args__ = ( db.CheckConstraint( "confidence >= 0 AND confidence <= 1", name="ck_candidate_confidence", ), db.CheckConstraint( "status IN ('candidate','accepted','rejected','ignored')", name="ck_candidate_status", ), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) job_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ingestion_jobs.uid", ondelete="CASCADE"), nullable=False, ) candidate_type = db.Column(db.String(40), nullable=False) normalized_data = db.Column(JSONB, nullable=False) evidence_uids = db.Column(JSONB, nullable=False, default=list) confidence = db.Column(db.Float, nullable=False) status = db.Column(db.String(20), nullable=False, default="candidate") created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) def to_dict(self) -> dict[str, Any]: return { "uid": str(self.uid), "job_uid": str(self.job_uid), "candidate_type": self.candidate_type, "normalized_data": dict(self.normalized_data or {}), "evidence_uids": list(self.evidence_uids or []), "confidence": float(self.confidence), "status": self.status, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), } class DataElement(db.Model): __tablename__ = "data_elements" __table_args__ = ( db.CheckConstraint( "status IN ('draft','in_review','published','deprecated','retired')", name="ck_data_element_status", ), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) code = db.Column(db.String(120), nullable=False, unique=True) current_version = db.Column(db.Integer, nullable=False, default=1) status = db.Column(db.String(20), nullable=False, default="draft") business_domain_uids = db.Column(JSONB, nullable=False, default=list) created_by = db.Column(db.String(100)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) versions = db.relationship( "DataElementVersion", lazy="selectin", order_by="DataElementVersion.version", cascade="all, delete-orphan", ) class DataElementVersion(db.Model): __tablename__ = "data_element_versions" __table_args__ = ( db.UniqueConstraint( "data_element_uid", "version", name="uq_data_element_version", ), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) data_element_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.data_elements.uid"), nullable=False, ) version = db.Column(db.Integer, nullable=False) status = db.Column(db.String(20), nullable=False) snapshot = db.Column(JSONB, nullable=False) evidence_uids = db.Column(JSONB, nullable=False, default=list) created_by = db.Column(db.String(100)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) class CandidateDecisionRecord(db.Model): __tablename__ = "candidate_decisions" __table_args__ = ( db.CheckConstraint( "action IN ('reuse','create','map','ignore')", name="ck_candidate_decision_action", ), db.UniqueConstraint("candidate_uid", name="uq_candidate_decision"), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) candidate_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.extraction_candidates.uid"), nullable=False, ) action = db.Column(db.String(20), nullable=False) data_element_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.data_elements.uid"), ) evidence_uids = db.Column(JSONB, nullable=False, default=list) payload = db.Column(JSONB, nullable=False, default=dict) actor_uid = db.Column(db.String(100)) reason = db.Column(db.String(1000)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) class OntologyModel(db.Model): __tablename__ = "ontologies" __table_args__ = ({"schema": "public"},) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) code = db.Column(db.String(120), nullable=False, unique=True) name = db.Column(db.String(300), nullable=False) owner_uid = db.Column(db.String(100), nullable=False) status = db.Column(db.String(20), nullable=False, default="draft") draft_revision = db.Column(db.Integer, nullable=False, default=0) active_version_uid = db.Column(UUID(as_uuid=False)) created_by = db.Column(db.String(100)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) class OntologyVersionModel(db.Model): __tablename__ = "ontology_versions" __table_args__ = ( db.UniqueConstraint("ontology_uid", "version", name="uq_ontology_version"), {"schema": "public"}, ) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) ontology_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), nullable=False ) version = db.Column(db.Integer, nullable=False) parent_version_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ontology_versions.uid") ) status = db.Column(db.String(20), nullable=False, default="draft") graph_document = db.Column(JSONB, nullable=False) content_hash = db.Column(db.String(64), nullable=False) created_by = db.Column(db.String(100)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) published_at = db.Column(db.DateTime) class OntologyDomainLinkModel(db.Model): __tablename__ = "ontology_domain_links" __table_args__ = ({"schema": "public"},) ontology_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), primary_key=True ) domain_uid = db.Column(UUID(as_uuid=False), primary_key=True) role = db.Column(db.String(20), nullable=False) class OntologyChangeSetModel(db.Model): __tablename__ = "ontology_change_sets" __table_args__ = ({"schema": "public"},) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) ontology_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), nullable=False ) base_version_uid = db.Column(UUID(as_uuid=False)) status = db.Column(db.String(20), nullable=False, default="draft") changes = db.Column(JSONB, nullable=False, default=list) decisions = db.Column(JSONB, nullable=False, default=list) created_by = db.Column(db.String(100)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) class OntologyPublishRunModel(db.Model): __tablename__ = "ontology_publish_runs" __table_args__ = ({"schema": "public"},) uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid) ontology_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), nullable=False ) version_uid = db.Column( UUID(as_uuid=False), db.ForeignKey("public.ontology_versions.uid"), nullable=False ) idempotency_key = db.Column(db.String(128), nullable=False, unique=True) status = db.Column(db.String(20), nullable=False) validation_result = db.Column(JSONB, nullable=False, default=dict) actor_uid = db.Column(db.String(100)) created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive) finished_at = db.Column(db.DateTime)