From 312778d995ee78e32cb1c5fd4ed9c29a063b7540 Mon Sep 17 00:00:00 2001 From: shangfangjian Date: Thu, 17 Sep 2026 01:15:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=95=B0=E6=8D=AE=E7=AE=A1=E7=BA=BF?= =?UTF-8?q?=E6=9E=B6=E6=9E=84=E4=BC=98=E5=8C=96=20+=20=E5=89=8D=E7=AB=AF?= =?UTF-8?q?=E9=83=A8=E7=BD=B2=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Bronze 层 RawEvent 表(原始事件存档) - IngestFailure 死信表(失败持久化+重试) - DataQualityCheck 数据质量监控 - DataLineage 血缘追踪 - MatchStats xG 追踪字段 - Nginx 前端服务配置 --- .../0008_raw_event_and_ingest_failure.py | 107 ++++++++++++++++++ .../versions/0009_match_stats_xg_fields.py | 29 +++++ frontend/nginx.conf | 38 +++++++ 3 files changed, 174 insertions(+) create mode 100644 alembic/versions/0008_raw_event_and_ingest_failure.py create mode 100644 alembic/versions/0009_match_stats_xg_fields.py create mode 100644 frontend/nginx.conf diff --git a/alembic/versions/0008_raw_event_and_ingest_failure.py b/alembic/versions/0008_raw_event_and_ingest_failure.py new file mode 100644 index 0000000..2a7544a --- /dev/null +++ b/alembic/versions/0008_raw_event_and_ingest_failure.py @@ -0,0 +1,107 @@ +"""创建 Bronze 层、死信表、质量监控、血缘追踪 4 张新表 + +Revision ID: 0008_raw_event_and_ingest_failure +Revises: 0007_predictions_unique_constraint +Create Date: 2026-09-17 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + +revision: str = '0008_raw_event_and_ingest_failure' +down_revision: Union[str, None] = '0007_predictions_unique_constraint' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # 1. RawEvent - Bronze 层原始事件存档 + op.create_table( + 'raw_events', + sa.Column('id', sa.BigInteger(), primary_key=True), + sa.Column('source_system', sa.String(50), nullable=False), + sa.Column('source_record_id', sa.String(100), nullable=False), + sa.Column('raw_payload', JSONB(), nullable=False), + sa.Column('ingested_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column('ingest_batch_id', sa.String(36), nullable=True), + sa.UniqueConstraint('source_system', 'source_record_id', name='uq_raw_event'), + ) + op.create_index('ix_raw_event_batch', 'raw_events', ['ingest_batch_id']) + + # 2. IngestFailure - 采集失败死信表 + op.create_table( + 'ingest_failures', + sa.Column('id', sa.BigInteger(), primary_key=True), + sa.Column('source_system', sa.String(50), nullable=False), + sa.Column('entity_type', sa.String(50), nullable=False), + sa.Column('source_record_id', sa.String(100), nullable=True), + sa.Column('error_type', sa.String(50), nullable=False), + sa.Column('error_detail', sa.Text(), nullable=True), + sa.Column('raw_payload', JSONB(), nullable=True), + sa.Column('retry_count', sa.Integer(), server_default='0'), + sa.Column('next_retry_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('status', sa.String(20), server_default='pending'), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column('resolved_at', sa.DateTime(timezone=True), nullable=True), + ) + op.create_index('ix_ingest_failure_status', 'ingest_failures', ['status', 'next_retry_at']) + op.create_check_constraint( + 'ck_ingest_failure_status', + 'ingest_failures', + "status IN ('pending', 'retrying', 'resolved', 'abandoned')", + ) + + # 3. DataQualityCheck - 数据质量监控 + op.create_table( + 'data_quality_checks', + sa.Column('id', sa.BigInteger(), primary_key=True), + sa.Column('check_name', sa.String(100), nullable=False), + sa.Column('entity_type', sa.String(50), nullable=False), + sa.Column('entity_id', sa.Integer(), nullable=True), + sa.Column('expected_value', sa.Float(), nullable=True), + sa.Column('actual_value', sa.Float(), nullable=False), + sa.Column('passed', sa.Boolean(), nullable=False), + sa.Column('severity', sa.String(10), server_default='warning'), + sa.Column('detail', JSONB(), nullable=True), + sa.Column('checked_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + op.create_index('ix_dqc_checked_at', 'data_quality_checks', ['checked_at']) + op.create_index('ix_dqc_entity', 'data_quality_checks', ['entity_type', 'entity_id']) + + # 4. DataLineage - ETL 血缘追踪 + op.create_table( + 'data_lineage', + sa.Column('id', sa.BigInteger(), primary_key=True), + sa.Column('source_system', sa.String(50), nullable=False), + sa.Column('source_record_id', sa.String(100), nullable=False), + sa.Column('target_table', sa.String(50), nullable=False), + sa.Column('target_id', sa.Integer(), nullable=True), + sa.Column('transform_name', sa.String(50), nullable=False), + sa.Column('transform_detail', JSONB(), nullable=True), + sa.Column('batch_id', sa.String(36), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + op.create_index('ix_lineage_source', 'data_lineage', ['source_system', 'source_record_id']) + op.create_index('ix_lineage_target', 'data_lineage', ['target_table', 'target_id']) + op.create_index('ix_lineage_batch', 'data_lineage', ['batch_id']) + + +def downgrade() -> None: + op.drop_index('ix_lineage_batch', table_name='data_lineage') + op.drop_index('ix_lineage_target', table_name='data_lineage') + op.drop_index('ix_lineage_source', table_name='data_lineage') + op.drop_table('data_lineage') + + op.drop_index('ix_dqc_entity', table_name='data_quality_checks') + op.drop_index('ix_dqc_checked_at', table_name='data_quality_checks') + op.drop_table('data_quality_checks') + + op.drop_index('ix_ingest_failure_status', table_name='ingest_failures') + op.drop_table('ingest_failures') + + op.drop_index('ix_raw_event_batch', table_name='raw_events') + op.drop_table('raw_events') diff --git a/alembic/versions/0009_match_stats_xg_fields.py b/alembic/versions/0009_match_stats_xg_fields.py new file mode 100644 index 0000000..c78007f --- /dev/null +++ b/alembic/versions/0009_match_stats_xg_fields.py @@ -0,0 +1,29 @@ +"""为 MatchStats 添加 xG 追踪字段 + +Revision ID: 0009_match_stats_xg_fields +Revises: 0008_raw_event_and_ingest_failure +Create Date: 2026-09-17 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = '0009_match_stats_xg_fields' +down_revision: Union[str, None] = '0008_raw_event_and_ingest_failure' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column('match_stats', sa.Column('xg_source', sa.String(30), nullable=True)) + op.add_column('match_stats', sa.Column('xg_updated_at', sa.DateTime(timezone=True), nullable=True)) + op.add_column('match_stats', sa.Column('xg_source_record_id', sa.String(100), nullable=True)) + + +def downgrade() -> None: + op.drop_column('match_stats', 'xg_source_record_id') + op.drop_column('match_stats', 'xg_updated_at') + op.drop_column('match_stats', 'xg_source') diff --git a/frontend/nginx.conf b/frontend/nginx.conf new file mode 100644 index 0000000..f1bb38b --- /dev/null +++ b/frontend/nginx.conf @@ -0,0 +1,38 @@ +server { + listen 80; + server_name localhost; + + # Serve static files + root /usr/share/nginx/html; + index index.html; + + # API proxy + location /api/ { + proxy_pass http://api:8000/api/; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection 'upgrade'; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_cache_bypass $http_upgrade; + } + + # Health check proxy + location /health { + proxy_pass http://api:8000/health; + proxy_http_version 1.1; + proxy_set_header Host $host; + } + + # SPA routing - serve index.html for all routes + location / { + try_files $uri $uri/ /index.html; + } + + # Gzip compression + gzip on; + gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript; + gzip_min_length 1000; +}