yuxuanhui
42 lines
1.7 KiB
Python
42 lines
1.7 KiB
Python
"""Classify snapshots whose sole Alpha failure is the unopened PPAC theme."""
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
|
|
revision = "0022"
|
|
down_revision = "0021"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def upgrade():
|
|
# Reuse the indexed check_type column; preserve all raw evidence and check stages.
|
|
table = sa.table("alphas", sa.column("id", sa.String()), sa.column("checks", sa.JSON()),
|
|
sa.column("check_type", sa.String(20)))
|
|
connection = op.get_bind()
|
|
last_id = None
|
|
while True:
|
|
query = sa.select(table.c.id, table.c.checks).order_by(table.c.id).limit(500)
|
|
if last_id is not None:
|
|
query = query.where(table.c.id > last_id)
|
|
rows = connection.execute(query).mappings().all()
|
|
if not rows:
|
|
break
|
|
candidates = []
|
|
for row in rows:
|
|
# Freeze the current interpretation instead of importing application code.
|
|
checks = row["checks"] if isinstance(row["checks"], list) else []
|
|
failures = [check for check in checks if isinstance(check, dict)
|
|
and check.get("name") != "REGULAR_SUBMISSION"
|
|
and isinstance(check.get("result"), str) and check["result"].upper() == "FAIL"]
|
|
if len(failures) == 1 and failures[0].get("name") == "PURE_POWER_POOL_THEME":
|
|
candidates.append(row["id"])
|
|
if candidates:
|
|
connection.execute(table.update().where(table.c.id.in_(candidates)).values(check_type="PPAC_CANDIDATE"))
|
|
last_id = rows[-1]["id"]
|
|
|
|
|
|
def downgrade():
|
|
table = sa.table("alphas", sa.column("check_type", sa.String(20)))
|
|
op.execute(table.update().where(table.c.check_type == "PPAC_CANDIDATE").values(check_type="FAIL_1"))
|