import re
import uuid
from typing import Sequence

from sqlalchemy import select, update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.subject import Subject
from app.models.subject_pd import SubjectPd
from app.schemas.subject_pd import SubjectPdCreate, SubjectPdUpdate


_PD_NO_PATTERN = re.compile(r"PD(\d+)$", re.IGNORECASE)


def _extract_pd_sequence(pd_no: str | None) -> int:
    if not pd_no:
        return 0
    matched = _PD_NO_PATTERN.search(str(pd_no).strip())
    if not matched:
        return 0
    try:
        return int(matched.group(1))
    except ValueError:
        return 0


async def _next_pd_no(db: AsyncSession, study_id: uuid.UUID) -> str:
    rows = await db.execute(select(SubjectPd.pd_no).where(SubjectPd.study_id == study_id))
    max_seq = 0
    for (pd_no,) in rows.all():
        max_seq = max(max_seq, _extract_pd_sequence(pd_no))
    return f"PD{max_seq + 1:04d}"


async def list_subject_pds(
    db: AsyncSession,
    study_id: uuid.UUID,
    subject_id: uuid.UUID,
    *,
    skip: int = 0,
    limit: int = 200,
) -> Sequence[SubjectPd]:
    stmt = (
        select(SubjectPd)
        .where(SubjectPd.study_id == study_id, SubjectPd.subject_id == subject_id)
        .order_by(SubjectPd.pd_no.asc(), SubjectPd.created_at.asc())
        .offset(skip)
        .limit(limit)
    )
    rows = await db.execute(stmt)
    return rows.scalars().all()


async def list_study_subject_pds(
    db: AsyncSession,
    study_id: uuid.UUID,
    *,
    skip: int = 0,
    limit: int = 500,
    site_ids: set[uuid.UUID] | None = None,
) -> list[tuple[SubjectPd, str, uuid.UUID]]:
    if site_ids is not None and not site_ids:
        return []
    stmt = (
        select(SubjectPd, Subject.subject_no, Subject.site_id)
        .join(Subject, Subject.id == SubjectPd.subject_id)
        .where(
            SubjectPd.study_id == study_id,
            Subject.study_id == study_id,
        )
        .order_by(SubjectPd.pd_no.asc(), SubjectPd.created_at.asc())
        .offset(skip)
        .limit(limit)
    )
    if site_ids is not None:
        stmt = stmt.where(Subject.site_id.in_(site_ids))
    rows = await db.execute(stmt)
    return [(row[0], row[1], row[2]) for row in rows.all()]


async def get_subject_pd(db: AsyncSession, pd_id: uuid.UUID) -> SubjectPd | None:
    row = await db.execute(select(SubjectPd).where(SubjectPd.id == pd_id))
    return row.scalar_one_or_none()


async def create_subject_pd(
    db: AsyncSession,
    study_id: uuid.UUID,
    pd_in: SubjectPdCreate,
    *,
    created_by: uuid.UUID | None = None,
) -> SubjectPd:
    payload = pd_in.model_dump(exclude_unset=True)
    # PD编号统一由后端按项目全局顺序生成，避免被外部覆盖。
    payload.pop("pd_no", None)
    pd_no = await _next_pd_no(db, study_id)
    item = SubjectPd(study_id=study_id, pd_no=pd_no, created_by=created_by, **payload)
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item


async def update_subject_pd(db: AsyncSession, item: SubjectPd, pd_in: SubjectPdUpdate) -> SubjectPd:
    update_data = pd_in.model_dump(exclude_unset=True)
    if update_data:
        await db.execute(
            sa_update(SubjectPd)
            .where(SubjectPd.id == item.id)
            .values(**update_data)
        )
        await db.commit()
    await db.refresh(item)
    return item


async def delete_subject_pd(db: AsyncSession, item: SubjectPd) -> None:
    await db.delete(item)
    await db.commit()