import json
from dataclasses import replace

import pytest

from edu_crawler.connectors import CONNECTOR_TYPES, connector_for
from edu_crawler.errors import SourceStructureChanged
from edu_crawler.models import ParsedItem
from edu_crawler.schema import validate_record


def parsed(value: dict, *, payload_hash: str = "a" * 64) -> ParsedItem:
    return ParsedItem(
        value=value,
        raw_snapshot_id="RAW-BOSS-20260824-ABCDEF012345",
        source_payload_hash=payload_hash,
        collected_at="2026-08-24T10:00:00+08:00",
        source_url="file:///authorized.ndjson",
    )


def test_all_nine_sources_have_concrete_connector_types() -> None:
    assert set(CONNECTOR_TYPES) == {
        "BOSS",
        "ZHAOPIN",
        "BRZP",
        "NCSS",
        "MOHRSS",
        "IGUOPIN",
        "HZRC",
        "SCHOOL_EMPLOYMENT",
        "SCHOOL_INTERNSHIP",
    }
    assert all(
        connector_for(code).describe()["source_code"] == code
        for code in CONNECTOR_TYPES
    )


def test_external_record_normalizes_and_redacts_contact() -> None:
    connector = connector_for("BOSS")
    source = {
        "source_record_id": "BOSS-001",
        "source_url": "https://www.bosszhipin.com/job_detail/BOSS-001.html",
        "title": "云原生开发工程师",
        "description": "<p>负责容器平台开发，联系 13800138000</p>",
        "company_name": "示例数字科技有限公司",
        "region_code": "330100",
        "employment_type": "全职",
        "published_at": "2026-08-23T08:00:00+08:00",
        "salary_min": "12000",
        "salary_max": "18000",
        "salary_period": "month",
        "skills": ["Docker", "Kubernetes", "Docker"],
    }
    record = connector.normalize(
        parsed(source),
        task_id="JOB-20260824-BOSS-001",
        data_version="V2026.08.24",
        default_region_code=None,
    )
    validate_record(connector.policy.target_schema, record)
    assert record["source_platform"] == "BOSS"
    assert record["job_description"] == "负责容器平台开发，联系 [CONTACT_REDACTED]"
    assert record["skills"] == ["Docker", "Kubernetes"]
    assert record["source_payload_hash"] == "a" * 64


def test_annual_salary_is_normalized_to_monthly_and_unknown_status_stays_unknown() -> (
    None
):
    connector = connector_for("BOSS")
    source = {
        "source_record_id": "BOSS-ANNUAL-001",
        "source_url": "https://www.bosszhipin.com/job_detail/BOSS-ANNUAL-001.html",
        "title": "年度薪资测试岗位",
        "description": "用于验证统一月薪折算",
        "company_name": "示例企业",
        "region_code": "330100",
        "salary_min": 120000,
        "salary_max": 240000,
        "salary_period": "year",
        "status": "new-upstream-code",
    }
    record = connector.normalize(
        parsed(source),
        task_id="JOB-20260824-BOSS-ANNUAL",
        data_version="V2026.08.24",
        default_region_code=None,
    )
    assert record["salary_min"] == 10000
    assert record["salary_max"] == 20000
    assert record["salary_period"] == "year"
    assert record["status"] == "unknown"


def test_content_hash_ignores_collection_metadata() -> None:
    connector = connector_for("BOSS")
    source = {
        "source_record_id": "BOSS-001",
        "source_url": "https://www.bosszhipin.com/job_detail/BOSS-001.html",
        "title": "云原生开发工程师",
        "description": "负责容器平台开发",
        "company_name": "示例数字科技有限公司",
        "region_code": "330100",
    }
    first = connector.normalize(
        parsed(source, payload_hash="a" * 64),
        task_id="JOB-20260824-BOSS-001",
        data_version="V2026.08.24",
        default_region_code=None,
    )
    second_item = replace(
        parsed(source, payload_hash="b" * 64),
        raw_snapshot_id="RAW-BOSS-20260825-012345ABCDEF",
        collected_at="2026-08-25T10:00:00+08:00",
    )
    second = connector.normalize(
        second_item,
        task_id="JOB-20260825-BOSS-001",
        data_version="V2026.08.25",
        default_region_code=None,
    )
    assert first["content_hash"] == second["content_hash"]
    assert first["source_payload_hash"] != second["source_payload_hash"]


def test_source_record_url_must_match_source_allowlist() -> None:
    connector = connector_for("BOSS")
    source = {
        "source_record_id": "BOSS-001",
        "source_url": "https://evil.example/job/1",
        "title": "测试岗位",
        "description": "测试职责",
        "company_name": "测试企业",
        "region_code": "330100",
    }
    try:
        connector.normalize(
            parsed(source),
            task_id="JOB-20260824-BOSS-001",
            data_version="V2026.08.24",
            default_region_code=None,
        )
    except Exception as exc:
        assert getattr(exc, "code", None) == "SOURCE_STRUCTURE_CHANGED"
    else:
        raise AssertionError("unapproved source URL was accepted")


def test_declared_source_must_match_selected_connector() -> None:
    connector = connector_for("BOSS")
    source = {
        "source_platform": "ZHAOPIN",
        "source_record_id": "ZL-001",
        "source_url": "https://www.bosszhipin.com/job_detail/1.html",
        "title": "测试岗位",
        "description": "测试岗位职责",
        "company_name": "测试企业",
        "region_code": "330100",
    }
    with pytest.raises(SourceStructureChanged) as caught:
        connector.normalize(
            parsed(source),
            task_id="JOB-20260824-BOSS-001",
            data_version="V2026.08.24",
            default_region_code=None,
        )
    assert getattr(caught.value, "code", None) == "SOURCE_STRUCTURE_CHANGED"


@pytest.mark.parametrize(
    ("source_code", "source_url"),
    [
        ("BOSS", "https://www.bosszhipin.com/job_detail/1.html"),
        ("ZHAOPIN", "https://www.zhaopin.com/jobdetail/1.htm"),
        ("BRZP", "https://brzp.chinasme.cn/brzp/job/1"),
        ("NCSS", "https://www.ncss.cn/student/jobs/1"),
        ("MOHRSS", "http://job.mohrss.gov.cn/cjobs/jobinfo/1"),
        ("IGUOPIN", "https://job.iguopin.com/job/detail?id=1"),
        ("HZRC", "https://www.hzrc.com/job/1"),
    ],
)
def test_each_external_source_normalizes_authorized_feed(
    source_code: str, source_url: str
) -> None:
    connector = connector_for(source_code)
    record = connector.normalize(
        parsed(
            {
                "source_record_id": f"{source_code}-001",
                "source_url": source_url,
                "title": "测试岗位",
                "description": "测试岗位职责",
                "company_name": "测试企业",
                "region_code": "330100",
            }
        ),
        task_id=f"JOB-20260824-{source_code}-001",
        data_version="V2026.08.24",
        default_region_code=None,
    )
    validate_record(connector.policy.target_schema, record)
    assert record["source_platform"] == source_code


def test_source_url_secrets_are_redacted() -> None:
    connector = connector_for("BOSS")
    record = connector.normalize(
        parsed(
            {
                "source_record_id": "BOSS-001",
                "source_url": "https://www.bosszhipin.com/job_detail/1.html?token=secret&city=330100#private",
                "title": "测试岗位",
                "description": "测试岗位职责",
                "company_name": "测试企业",
                "region_code": "330100",
            }
        ),
        task_id="JOB-20260824-BOSS-001",
        data_version="V2026.08.24",
        default_region_code=None,
    )
    assert "secret" not in record["source_url"]
    assert "private" not in record["source_url"]
    assert "city=330100" in record["source_url"]


def test_schema_org_job_posting_mapping() -> None:
    connector = connector_for("BRZP")
    source = {
        "@type": "JobPosting",
        "identifier": {"value": "BRZP-001"},
        "url": "https://brzp.chinasme.cn/brzp/job/BRZP-001",
        "title": "跨境电商运营实习生",
        "description": "协助店铺运营和数据分析",
        "hiringOrganization": {"name": "示例跨境服务有限公司"},
        "jobLocation": {"address": {"addressLocality": "杭州市"}},
        "employmentType": "internship",
        "datePosted": "2026-08-23T08:00:00+08:00",
        "baseSalary": {
            "currency": "CNY",
            "value": {"minValue": 3500, "maxValue": 5000, "unitText": "MONTH"},
        },
    }
    record = connector.normalize(
        parsed(source),
        task_id="JOB-20260824-BRZP-001",
        data_version="V2026.08.24",
        default_region_code="330100",
    )
    validate_record(connector.policy.target_schema, record)
    assert record["employment_type"] == "internship"
    assert record["salary_period"] == "month"
    assert json.loads(json.dumps(record, ensure_ascii=False)) == record