CoolBotDataSys/alembic/env.py

from logging.config import fileConfig
from sqlalchemy import engine_from_config, MetaData, Table, Column, String, Integer, BigInteger, Boolean, Text, Float, DateTime, Index
from sqlalchemy import pool
from alembic import context
import sys
import os

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

config = context.config

# 数据库配置
db_config = {
    'host': 'pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com',
    'port': 5432,
    'user': 'coolbot',
    'password': 'Coolbot123',
    'database': 'coolbot_data'
}

config.set_main_option('sqlalchemy.url', 
    f"postgresql://{db_config['user']}:{db_config['password']}@"
    f"{db_config['host']}:{db_config['port']}/{db_config['database']}")

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# 定义 metadata（用于 autogenerate）
metadata = MetaData()

# 定义现有表结构
Table('collections', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('category', String(100)),
    Column('series', String(100)),
    Column('issue_year', Integer),
    Column('description', Text),
    Column('image_url', String(500)),
    Column('created_at', DateTime),
    Column('updated_at', DateTime),
)

Table('yichens_posts', metadata,
    Column('id', BigInteger, primary_key=True, autoincrement=True),
    Column('post_id', String(100), unique=True),
    Column('topic_id', String(100)),
    Column('title', String(500)),
    Column('content', Text),
    Column('author_id', String(100)),
    Column('author_username', String(100)),
    Column('category', String(50)),
    Column('sub_category', String(50)),
    Column('post_type', String(20)),
    Column('price', Float),
    Column('price_unit', String(20)),
    Column('contact', String(200)),
    Column('special_types', String(200)),
    Column('has_lifebuoy', Boolean),
    Column('reply_count', Integer),
    Column('view_count', Integer),
    Column('post_url', String(500)),
    Column('created_at', DateTime),
    Column('updated_at', DateTime),
    Column('crawled_at', DateTime),
    Column('url', String(500)),
    Index('idx_yichens_posts_post_id', 'post_id'),
    Index('idx_yichens_posts_category', 'category'),
    Index('idx_yichens_posts_post_time', 'created_at'),
)

Table('price_history', metadata,
    Column('id', BigInteger, primary_key=True, autoincrement=True),
    Column('collection_id', Integer),
    Column('price', Float),
    Column('price_unit', String(20)),
    Column('source', String(50)),
    Column('url', String(500)),
    Column('post_type', String(20)),
    Column('special_types', String(200)),
    Column('author', String(100)),
    Column('contact', String(200)),
    Column('content', Text),
    Column('crawled_at', DateTime),
    Column('created_at', DateTime),
)

Table('crawl_logs', metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('source', String(50)),
    Column('status', String(20)),
    Column('items_count', Integer),
    Column('error_message', Text),
    Column('started_at', DateTime),
    Column('finished_at', DateTime),
)

Table('yichens_members', metadata,
    Column('user_id', String(100), primary_key=True),
    Column('username', String(100), unique=True),
    Column('transaction_level', String(50)),
    Column('credit_score', Integer),
    Column('rating_count', Integer),
    Column('post_count', Integer),
    Column('post_points', Integer),
    Column('has_business_license', Boolean),
    Column('license_info', Text),
    Column('identity_verified', Boolean),
    Column('real_name', String(100)),
    Column('verification_notes', Text),
    Column('phone', String(100)),
    Column('address', String(500)),
    Column('bank_accounts', Text),
    Column('alipay', String(200)),
    Column('registration_date', DateTime),
    Column('member_since', String(100)),
    Column('is_verified', Boolean),
    Column('status', String(20)),
    Column('created_at', DateTime),
)

target_metadata = metadata

def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection, 
            target_metadata=target_metadata
        )
        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()