from __future__ import annotations

import ast
import json
import os
import re
import shutil
from pathlib import Path

ROOT = Path(__file__).resolve().parent
SOURCE = (ROOT / 'ws8771_main.py').read_text(encoding='utf-8')
CONFIG = json.loads((ROOT / 'ws8771_config.json').read_text(encoding='utf-8'))


def test_ws_version_and_direct_finalize_handlers() -> None:
    assert 'WS_BUILD_VERSION = "2.3.8.250"' in SOURCE
    assert 'WS_BUILD_TAG = "v250"' in SOURCE
    assert 'WS_LABEL = "WS v250"' in SOURCE
    assert 'ws8771_intentional_stop.json' not in SOURCE
    assert 'write_ws_intentional_stop_marker' not in SOURCE
    assert 'def handle_ig_media_exists' in SOURCE
    assert 'def handle_ig_media_download_direct' in SOURCE
    assert 'def handle_ig_media_downloaded' in SOURCE
    assert 'def generate_account_html_report' in SOURCE
    assert 'def handle_ig_report_rebuild_all' in SOURCE
    assert 'if act == "ig_report_rebuild_all"' in SOURCE



def test_hourly_central_statistics_sync_handler() -> None:
    for token in [
        'def handle_ig_observation_sync_hour(self, req):',
        '"hour_bucket": hour_bucket',
        '"share_count": 0',
        'yellow_api_call(self.cfg, "media_stat_hour_upsert"',
        'yellow_api_call(self.cfg, "media_stat_hour_list"',
        '통계 1시간 연동 account=',
        'if act == "ig_observation_sync_hour"',
    ]:
        assert token in SOURCE, token
    actions = set((CONFIG.get('thread_queue_mode') or {}).get('actions') or [])
    assert 'ig_observation_sync_hour' in actions


def test_physical_mouse_modules_are_removed() -> None:
    for name in [
        'ws8771_ig_auto_click_download.py',
        'ws8771_ig_auto_download.py', 'ws8771_origin_crosscheck.py'
    ]:
        assert not (ROOT / name).exists(), name
    forbidden = [
        'MOUSEEVENTF_LEFTDOWN', 'mouse_left_click(', 'mouse_move_smooth(',
        'SetCursorPos', 'GetCursorPos', 'WindowFromPoint',
        'ig_rtc_auto_click_once', 'ig_rtc_mouse_sync', 'ig_rtc_hover_status',
        'hert8772position', 'ig_dom_calibration'
    ]
    for token in forbidden:
        assert token not in SOURCE, token


def test_config_has_no_retired_mouse_state_and_auto_enter_is_enabled() -> None:
    for key in ['user_mouse_idle_resume_sec', 'user_mouse_move_threshold_px', 'ig_dom_calibration', 'hert8772position', 'mouse_action_rest_sec']:
        assert key not in CONFIG
    actions = set((CONFIG.get('thread_queue_mode') or {}).get('actions') or [])
    assert 'ig_rtc_auto_click_once' not in actions
    assert {'ig_media_exists', 'ig_media_download_direct', 'ig_media_downloaded', 'ig_observation_sync_hour', 'ig_report_rebuild_all'}.issubset(actions)
    assert 'def handle_ig_rtc_enter_auto_arm' in SOURCE
    assert CONFIG['enabled'] is True
    assert CONFIG['enter_mode'] == 'auto'
    assert CONFIG['try_enter_after_idok'] is True
    assert CONFIG['match_contains'] is True


def test_instagram_save_dialog_enter_flow_is_wired() -> None:
    for token in [
        'def instagram_download_dialog_match',
        'self.auto_enter_expected = {',
        'self.arm_auto_enter(reason, data)',
        'find_save_dialog_filename_edit(hwnd, expected)',
        'ig_match, ig_match_reason = instagram_download_dialog_match(filename, expected)',
        'if auto_unarmed and not ig_match:',
        'auto_accept_save_dialog(hwnd, self.cfg, expected)',
        'press_enter_key()',
    ]:
        assert token in SOURCE, token
    # Instagram 파일 패턴만 자동 확정하고 다른 프로그램 Save As는 건드리지 않는다.
    for token in ['_No000_', '_No없음_', 'ig_filename_pattern', 'filename_pattern_mismatch']:
        assert token in SOURCE, token

    tree = ast.parse(SOURCE)
    node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'instagram_download_dialog_match')
    scope = {'os': os, 're': re}
    exec(compile(ast.Module(body=[node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)
    match = scope['instagram_download_dialog_match']
    assert match('owner_No000_Da0WTq-OosH.mp4', {}) == (True, 'ig_filename_pattern')
    assert match('owner_No003_Da0WTq-OosH.jpg', {'shortcode': 'Da0WTq-OosH'}) == (True, 'expected_shortcode')
    assert match('report.xlsx', {}) == (False, 'not_media_extension')
    assert match('vacation.jpg', {}) == (False, 'filename_pattern_mismatch')


def test_fixed_stdown_instagram_account_storage(tmp_path: Path) -> None:
    for token in [
        'derive_stdown_root(ST_ROOT) / "instagram"',
        'def relocate_download_to_expected_folder',
        'download_relocate_failed',
        'target_path_enter',
        'enter_relocate_fallback',
        'write_control_text(edit_hwnd, target_path)',
    ]:
        assert token in SOURCE, token

    tree = ast.parse(SOURCE)
    node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'expected_download_folder')
    scope = {
        'Path': Path,
        'ST_ROOT': Path('/tmp/bundle/_St'),
        'derive_stdown_root': lambda root: Path(root).parent / '_StDown',
        'safe_path_part': lambda value, fallback='unknown': str(value or fallback),
    }
    exec(compile(ast.Module(body=[node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)
    expected = scope['expected_download_folder']({'account': 'account_a'})
    assert expected.replace('\\', '/').endswith('/_StDown/instagram/account_a')
    merged = scope['expected_download_folder']({'account': 'account_a', 'account_folder_merge': True})
    assert merged.replace('\\', '/').endswith('/_StDown/instagram/account_a')

    relocate_node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'relocate_download_to_expected_folder')
    target_root = tmp_path / '_StDown' / 'instagram' / 'account_a'
    relocate_scope = {
        'Path': Path,
        'os': os,
        'shutil': shutil,
        'expected_download_folder': lambda data: str(target_root),
        'file_hashes': lambda path: ('md5', 'sha256', Path(path).stat().st_size),
    }
    exec(compile(ast.Module(body=[relocate_node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), relocate_scope)
    source = tmp_path / 'Downloads' / 'owner_No001_CODE1.mp4'
    source.parent.mkdir(parents=True)
    source.write_bytes(b'media')
    moved_path, moved = relocate_scope['relocate_download_to_expected_folder'](source, {'account': 'account_a'})
    assert moved is True
    assert Path(moved_path) == target_root / source.name
    assert Path(moved_path).read_bytes() == b'media'
    assert not source.exists()


def test_ws_records_shared_media_saved_after_file_finalize() -> None:
    for token in [
        "from Shared.storage import canonical_instagram_post_url, derive_stdown_root, record_downloaded_file",
        "def record_shared_media_saved",
        "canonical_instagram_post_url(",
        "MEDIA_SAVED_OK",
        "storage_ok",
        "client_event_id",
    ]:
        assert token in SOURCE, token


def test_save_dialog_state_callback_exists_for_ui_diagnostics() -> None:
    assert 'def save_dialog_state(self, is_open, source, dialogs)' in SOURCE
    assert 'save_dialog_state' in SOURCE
    # v230 field omission caused repeated Tk callback exceptions; the callback must now be concrete.
    tree = ast.parse(SOURCE)
    floating = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'FloatingApp')
    assert any(isinstance(n, ast.FunctionDef) and n.name == 'save_dialog_state' for n in floating.body)


def test_ws_direct_download_streams_to_target(tmp_path: Path) -> None:
    assert 'def download_url_to_file' in SOURCE
    assert 'DIRECT_DOWNLOAD_BEGIN' in SOURCE
    assert 'DIRECT_DOWNLOAD_OK' in SOURCE

    tree = ast.parse(SOURCE)
    node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'download_url_to_file')
    import threading
    import urllib.request
    scope = {'Path': Path, 'os': os, 'threading': threading, 'urllib': __import__('urllib')}
    exec(compile(ast.Module(body=[node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)

    source = tmp_path / 'source.mp4'
    source.write_bytes(b'instagram-media-bytes')
    target = tmp_path / '_StDown' / 'instagram' / 'account_a' / 'source.mp4'
    result = scope['download_url_to_file'](source.as_uri(), target, headers={}, timeout=5)
    assert result['ok'] is True
    assert Path(result['file_path']) == target.resolve()
    assert target.read_bytes() == b'instagram-media-bytes'
    assert not list(target.parent.glob('*.part'))


def test_feature_check_log_is_korean_concise_and_deduplicated() -> None:
    for token in [
        'feature_check_mode = bool(data.get("feature_check_mode"))',
        'check_id = str(data.get("check_id") or "").strip()',
        'check_status = str(data.get("check_status") or "").strip()',
        'check_message_ko = str(data.get("check_message_ko") or "").strip()',
        'def project_progress_check(self, feature_name, check_id, status, message_ko, detail=None):',
        '_project_progress_seen',
        '_append_project_progress_line(judgement)',
        '검색어=*** 상세로그 ***',
        'ws_project_progress_log',
        'ws_failure_log',
        'elif (not feature_check_mode) and (stage or decision or action):',
    ]:
        assert token in SOURCE, token


def test_project_progress_and_failure_logs_are_linked(tmp_path: Path, capsys) -> None:
    tree = ast.parse(SOURCE)
    function_names = {
        'project_log_context_snapshot',
        '_append_project_progress_line',
        '_project_failure_keyword',
        '_write_project_failure_excerpt',
        '_write_run_detail_marker',
        '_prune_python_logs',
    }
    nodes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in function_names]
    floating = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'FloatingApp')
    method = next(n for n in floating.body if isinstance(n, ast.FunctionDef) and n.name == 'project_progress_check')

    import threading
    from datetime import datetime
    from types import SimpleNamespace

    progress_path = tmp_path / 'ws_project_progress_log_test.log'
    run_path = tmp_path / 'ws_run_log_test.log'
    progress_file = progress_path.open('a', encoding='utf-8', buffering=1)
    scope = {
        'threading': threading,
        'datetime': datetime,
        'os': os,
        '_PROJECT_LOG_LOCK': threading.RLock(),
        '_PROJECT_LOG_CONTEXT': {
            'enabled': True,
            'run_id': 'TEST_RUN',
            'run_log': str(run_path),
            'error_log': str(tmp_path / 'ws_err_log_test.log'),
            'project_progress_log': str(progress_path),
            'failure_dir': str(tmp_path),
            'failure_prefix': 'ws_failure_log',
        },
        '_PROJECT_PROGRESS_FILE': progress_file,
        'now_text': lambda: '2026-07-22 12:00:00',
        'safe_path_part': lambda value, fallback='x': re.sub(r'[^0-9A-Za-z._-]+', '_', str(value or fallback)).strip('_') or fallback,
    }
    exec(compile(ast.Module(body=nodes + [method], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)

    fake = SimpleNamespace()
    progress = scope['project_progress_check']
    progress(fake, '자동다운로드', 'AD-01', '[체크완료]', '자동 시작 활성화', {'cycle_id': 'run_1'})
    progress(fake, '자동다운로드', 'AD-01', '[체크완료]', '자동 시작 활성화', {'cycle_id': 'run_1'})
    progress(fake, '자동다운로드', 'AD-03', '[체크실패]', '대상 불일치', {'cycle_id': 'No001_CODE', 'error_code': 'target_mismatch'})
    progress(fake, '자동다운로드', 'AD-06', '[체크완료]', '파일 저장 완료', {'cycle_id': 'No001_CODE'})
    result = progress(fake, '서버등록', 'AD-06-S01', '[체크실패]', 'HTTP 500 / 다음점검=media_insert', {
        'cycle_id': 'No001_CODE',
        'shortcode': 'CODE',
        'asset_kind': 'video',
        'error_code': 'HTTP 500',
        'next_check': 'media_insert',
        'blocking': False,
    })
    progress_file.close()

    text = progress_path.read_text(encoding='utf-8')
    assert text.count('[자동다운로드][AD-01][체크완료]') == 1
    assert '[자동다운로드][AD-03][복구완료]' in text
    assert '[서버등록][AD-06-S01][체크실패]' in text
    assert '다음단계=로컬처리계속·실패항목별도추적' in text
    assert f'상세로그={run_path}' in text
    assert '검색어=*** 상세로그 *** PP-AD-06-S01-' in text
    failure_path = Path(result['failure_log'])
    assert failure_path.exists()
    failure_text = failure_path.read_text(encoding='utf-8')
    assert f'원본 상세로그: {run_path}' in failure_text
    assert '검색어: *** 상세로그 *** PP-AD-06-S01-' in failure_text
    captured = capsys.readouterr().out
    assert '*** 상세로그 *** PP-AD-06-S01-' in captured


def test_yellow_http_error_body_is_preserved(monkeypatch) -> None:
    import io
    import importlib.util
    import urllib.error

    spec = importlib.util.spec_from_file_location('ws8771_ig_sort_test', ROOT / 'ws8771_ig_sort.py')
    assert spec is not None and spec.loader is not None
    ws8771_ig_sort = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(ws8771_ig_sort)

    body = json.dumps({
        'ok': 0,
        'error': 'schema_mismatch',
        'error_stage': 'collect_event_schema',
        'table': 'ig_media_collect_event',
        'missing_columns': ['pass_id', 'rank_label'],
        'message': 'required columns are missing',
        'sql_error_code': 1054,
    }, ensure_ascii=False).encode('utf-8')

    def fail_urlopen(req, timeout=0):
        raise urllib.error.HTTPError(
            req.full_url,
            500,
            'Internal Server Error',
            hdrs=None,
            fp=io.BytesIO(body),
        )

    monkeypatch.setattr(ws8771_ig_sort.urllib.request, 'urlopen', fail_urlopen)
    logs = []
    result = ws8771_ig_sort.call_api(
        {'yellow_api': {'enabled': True, 'url': 'https://example.invalid/yellow.php', 'token': 'secret-token'}},
        'media_insert',
        {'shortcode': 'CODE1', 'asset_kind': 'video'},
        logger=logs.append,
    )

    assert result['ok'] is False
    assert result['http_status'] == 500
    assert result['error'] == 'schema_mismatch'
    assert result['error_stage'] == 'collect_event_schema'
    assert result['table'] == 'ig_media_collect_event'
    assert result['missing_columns'] == ['pass_id', 'rank_label']
    assert result['sql_error_code'] == 1054
    assert any('stage=collect_event_schema' in line for line in logs)
    assert all('secret-token' not in line for line in logs)


def test_server_failure_progress_contains_schema_diagnostics() -> None:
    for token in [
        '"http_status": detail_doc.get("http_status", "")',
        '"error_stage": detail_doc.get("error_stage", "")',
        '"server_message": detail_doc.get("server_message", "")',
        '"missing_columns": detail_doc.get("missing_columns", "")',
        '"http_status": yellow_res.get("http_status", "")',
        '"error_stage": str(yellow_res.get("error_stage") or "")',
    ]:
        assert token in SOURCE, token


def test_html_report_uses_published_time_fixed_thumbnail_and_full_view(tmp_path: Path) -> None:
    report_source = (ROOT / 'ws8771_html_report.py').read_text(encoding='utf-8')
    assert '<dt>게시 시각</dt>' in report_source
    assert '<dt>저장 시각</dt>' not in report_source
    assert 'item.get("upload_at") or item.get("published_at")' in report_source
    assert 'repeat(auto-fill,320px)' in report_source
    assert '.post-card{{width:320px' in report_source
    assert 'aspect-ratio:1/1' in report_source
    assert '>전체보기</a>' in report_source
    assert 'http://127.0.0.1:8780/' in report_source
    assert 'def _report_account_name' in report_source
    assert 'def rebuild_account_reports' in report_source
    assert 'REPORT_VERSION = 4' in report_source
    assert '"top_n": normalize_int(meta.get("top_n"), 0)' in SOURCE
    assert 'meta.get("top_n") or meta.get("rank_total")' not in SOURCE

    module_scope: dict[str, object] = {}
    exec(compile(report_source, str(ROOT / 'ws8771_html_report.py'), 'exec'), module_scope)
    display_time = module_scope['_display_time']
    assert display_time({'upload_at': '2026-07-20 12:34:56', 'downloaded_at': '2026-07-25 14:00:00'}) == '2026-07-20 12:34:56'
    assert display_time({'published_at': '2026-07-19 01:02:03', 'collected_at': '2026-07-25 14:00:00'}) == '2026-07-19 01:02:03'
    assert display_time({'downloaded_at': '2026-07-25 14:00:00'}) == ''


def test_report_rebuild_all_generates_account_specific_analyze_link(tmp_path: Path) -> None:
    import importlib.util

    module_path = ROOT / 'ws8771_html_report.py'
    spec = importlib.util.spec_from_file_location('ws8771_html_report_v249_test', module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    root = tmp_path / 'instagram'
    media = root / 'lilys_home'
    media.mkdir(parents=True)
    (media / 'sample.jpg').write_bytes(b'jpg')
    doc = {
        'folder_name': 'lilys_home',
        'items': {
            'CODE1|image|0': {
                'shortcode': 'CODE1',
                'asset_kind': 'image',
                'file_name': 'sample.jpg',
                'upload_at': '2026-07-20 12:34:56',
            }
        },
    }
    json_path = root / 'lilys_home.json'
    json_path.write_text(json.dumps(doc, ensure_ascii=False), encoding='utf-8')

    result = module.rebuild_account_reports(str(root))
    assert result['ok'] is True
    assert result['requested_count'] == 1
    assert result['success_count'] == 1
    html_text = (root / 'lilys_home.html').read_text(encoding='utf-8')
    legacy_text = (media / 'index.htm').read_text(encoding='utf-8')
    expected = 'http://127.0.0.1:8780/?account=lilys_home'
    assert expected in html_text
    assert expected in legacy_text


def test_observation_hour_sync_uses_atomic_hour_api() -> None:
    text = SOURCE
    for token in (
        'yellow_api_call(self.cfg, "media_stat_hour_upsert"',
        '"items_json": json.dumps(normalized',
        'yellow_api_call(self.cfg, "media_stat_hour_list"',
        'account_scope_mismatch',
        'saved_count != len(normalized)',
        'duplicate_shortcode',
        'missing_shortcodes = sorted(expected_shortcodes - received_shortcodes)',
    ):
        assert token in text, token


def test_injection_download_metadata_sort_fallback_and_merge_preservation() -> None:
    tree = ast.parse(SOURCE)
    wanted = {'download_meta_sort', 'download_meta_metric', 'merge_download_item_metadata'}
    nodes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in wanted]
    scope = {'normalize_int': lambda value, default=0: int(value or default)}
    exec(compile(ast.Module(body=nodes, type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)
    sort_payload = scope['download_meta_sort']({'sort_snapshot': {'views': 321, 'likes': 12, 'comments': 3, 'reposts': 2}})
    assert sort_payload['views'] == 321
    current = {'shortcode': 'CODE1', 'views': 0, 'likes': 0, 'comments': 0, 'reposts': 0, 'thumbnail_url': '', 'sort': {}}
    previous = {'shortcode': 'CODE1', 'views': 111, 'likes': 9, 'comments': 2, 'reposts': 1, 'thumbnail_url': 'old.jpg', 'sort': {'views': 111}}
    missing_meta = {'shortcode': 'CODE1', 'metric_presence': {'views': False, 'likes': False, 'comments': False, 'reposts': False}}
    merged = scope['merge_download_item_metadata'](previous, current, missing_meta)
    assert merged['views'] == 111
    assert merged['likes'] == 9
    assert merged['thumbnail_url'] == 'old.jpg'
    explicit = {'shortcode': 'CODE1', 'views': 0, 'likes': 0, 'comments': 0, 'reposts': 0, 'metric_presence': {'views': True, 'likes': True, 'comments': True, 'reposts': True}, 'sort': {'views': 0, 'likes': 0, 'comments': 0, 'reposts': 0}}
    merged_zero = scope['merge_download_item_metadata'](previous, current, explicit)
    assert merged_zero['views'] == 0
    assert merged_zero['likes'] == 0


def test_duplicate_metadata_refresh_merges_current_shortcode_stats() -> None:
    for token in [
        'incoming = {',
        'merged_item = merge_download_item_metadata(it, incoming, meta)',
        'download_meta_metric(meta, sort, metric_name)',
        'sort = download_meta_sort(meta)',
        '"caption": str(meta.get("caption") or "")',
    ]:
        assert token in SOURCE, token
