Generation

def img_label(item: Image | None) -> str:
    """Get the label to show in progress bars when iterating images."""
    if item is None:
        return ""
    return item.clean_name

def article_label(item: Article | None) -> str:
    """Get the label to show in progress bar when iterating articles."""
    if item is None:
        return ""
    return constraint(item.title, 25)

def constraint(value: str, limit: int) -> str:
    """Limit a string to a certain length if exceeded."""
    if len(value) <= limit:
        return value
    return value[: limit - 1] + "…"

def progress_bar(
    iterable: Iterable[V] | None = None,
    length: int | None = None,
    label: str | None = None,
    item_show_func: Callable[[V | None], str | None] | None = None,
    info_sep: str = "  ",
    width: int = 36,
) -> ProgressBar[V]:
    """Get a progress bar iterator."""
    return click.progressbar(
        iterable,
        length,
        label,
        True,
        True,
        True,
        item_show_func,
        "█",
        "░",
        f"%(label)s {Fore.YELLOW}%(bar)s{Style.RESET_ALL} %(info)s",
        info_sep,
        width,
    )

def fetch_category_entries(category: str, exclude_titles: set[str] | None = None) -> list[WikiEntry]:
    """Fetch a list of wiki entries in a certain category."""
    click.echo(f"Fetching articles in {Fore.BLUE}Category:{category}{Style.RESET_ALL}...")
    entries = []
    with timed() as t:
        for entry in wiki_client.get_category_members(category):
            if exclude_titles and entry.title in exclude_titles:
                continue
            if entry.title.startswith("User:") or entry.title.startswith("TibiaWiki:"):
                continue
            entries.append(entry)
    click.echo(f"\t{Fore.GREEN}Found {len(entries):,} articles in {t.elapsed:.2f} seconds.{Style.RESET_ALL}")
    return entries

def write_parsing_error(
    file: TextIO,
    *,
    category: str,
    article: Article,
    error: ArticleParsingError,
) -> None:
    """Write a parsing error entry to a log file."""
    file.write(f"{PARSING_ERROR_SEPARATOR}\n")
    file.write(f"Category: {category}\n")
    file.write(f"Article: {article.title}\n")
    file.write(f"URL: {article.url}\n")
    file.write("Traceback:\n")
    file.writelines(traceback.format_exception(type(error), error, error.__traceback__))
    file.write("\n")

def parse_articles(
    conn: sqlite3.Connection,
    data_store: dict[str, Any],
    enabled_categories: set[str],
    parsing_errors_log: TextIO | None = None,
) -> int:
    """Parse category articles into the database."""
    click.echo("Parsing articles...")
    parsing_errors_count = 0
    for key, category in CATEGORIES.items():
        if key not in enabled_categories:
            continue

        titles = [entry.title for entry in data_store[key]]
        parser = category.parser
        if category.generate_map:
            data_store[f"{key}_map"] = {}
        unparsed = []
        generator = wiki_client.get_articles(titles)
        with (
            timed() as t,
            conn,
            progress_bar(generator, len(titles), f"Parsing {key}", item_show_func=article_label) as bar,
        ):
            for article in bar:
                try:
                    entry = parser.from_article(article)
                    entry.insert(conn)
                    if category.generate_map:
                        data_store[f"{key}_map"][entry.title.lower()] = entry.article_id
                except ArticleParsingError as e:
                    unparsed.append(article.title)
                    parsing_errors_count += 1
                    if parsing_errors_log:
                        write_parsing_error(parsing_errors_log, category=key, article=article, error=e)
        if unparsed:
            click.echo(f"{Fore.RED}Could not parse {len(unparsed):,} articles.{Style.RESET_ALL}")
            click.echo(f"\t-> {Fore.RED}{f'{Style.RESET_ALL},{Fore.RED}'.join(unparsed)}{Style.RESET_ALL}")
        click.echo(f"\t{Fore.GREEN}Parsed articles in {t.elapsed:.2f} seconds.{Style.RESET_ALL}")
    return parsing_errors_count

def resolve_enabled_categories(skip_categories: set[str]) -> tuple[set[str], dict[str, set[str]]]:
    """Resolve enabled categories including dependency-based auto-skips."""
    enabled_categories = set(CATEGORIES).difference(skip_categories)
    auto_skipped: dict[str, set[str]] = {}
    changed = True
    while changed:
        changed = False
        for key, category in CATEGORIES.items():
            if key not in enabled_categories:
                continue
            missing_dependencies = {dep for dep in category.depends_on if dep not in enabled_categories}
            if not missing_dependencies:
                continue
            enabled_categories.remove(key)
            auto_skipped[key] = missing_dependencies
            changed = True
    return enabled_categories, auto_skipped

def warn_auto_skipped_categories(auto_skipped_categories: dict[str, set[str]]) -> None:
    """Emit warnings for categories that were disabled due to dependencies."""
    for key in CATEGORIES:
        if key not in auto_skipped_categories:
            continue
        dependencies = ", ".join(sorted(auto_skipped_categories[key]))
        click.echo(
            f"{Fore.YELLOW}Skipping category '{key}' because required categories are disabled: "
            f"{dependencies}.{Style.RESET_ALL}",
        )

def run_post_tasks(
    conn: sqlite3.Connection,
    data_store: dict[str, Any],
    enabled_categories: set[str],
    skip_images: bool,
) -> None:
    """Run post-processing tasks honoring dependency constraints."""
    for post_task in POST_TASKS:
        if post_task.name == "images" and skip_images:
            continue
        missing_dependencies = [dep for dep in post_task.dependencies if dep not in enabled_categories]
        if missing_dependencies:
            dependencies = ", ".join(sorted(missing_dependencies))
            click.echo(
                f"{Fore.YELLOW}Skipping task '{post_task.name}' because required categories are disabled: "
                f"{dependencies}.{Style.RESET_ALL}",
            )
            continue
        post_task.callback(conn, data_store, enabled_categories)

def generate(
    conn: sqlite3.Connection,
    skip_images: bool = False,
    skip_deprecated: bool = False,
    include_deprecated_images: bool = False,
    skip_categories: tuple[str, ...] = (),
    parsing_errors_file: str | None = None,
) -> None:
    """Generate a complete TibiaWiki SQLite database."""
    normalized_skip_categories = {category.casefold() for category in skip_categories}
    unknown_categories = normalized_skip_categories - set(CATEGORIES)
    if unknown_categories:
        unknown_str = ", ".join(sorted(unknown_categories))
        msg = f"Unknown categories in skip list: {unknown_str}."
        raise ValueError(msg)

    enabled_categories, auto_skipped_categories = resolve_enabled_categories(normalized_skip_categories)
    warn_auto_skipped_categories(auto_skipped_categories)

    click.echo("Creating schema...")
    schema.create_tables(conn)
    conn.execute("PRAGMA synchronous = OFF")
    data_store: dict[str, Any] = {}

    if skip_deprecated:
        deprecated = {
            entry.title
            for category in ("Deprecated", "Unavailable")
            for entry in fetch_category_entries(category)
        }
    else:
        deprecated = set()

    deprecated_image_titles: dict[str, list[str]] = {}

    for key, category in CATEGORIES.items():
        if key not in enabled_categories:
            continue
        excludes_deprecated = skip_deprecated and not category.include_deprecated
        category_has_images = not category.no_images or key == "outfits"
        if include_deprecated_images and not skip_images and excludes_deprecated and category_has_images:
            entries = fetch_category_entries(category.name)
            data_store[key] = [entry for entry in entries if entry.title not in deprecated]
            deprecated_image_titles[key] = [entry.title for entry in entries if entry.title in deprecated]
        else:
            excluded_titles = deprecated if excludes_deprecated else None
            data_store[key] = fetch_category_entries(category.name, excluded_titles)

    if deprecated_image_titles:
        data_store["deprecated_image_titles"] = deprecated_image_titles

    parsing_errors_path = Path(parsing_errors_file) if parsing_errors_file else None
    if parsing_errors_path:
        with parsing_errors_path.open("w", encoding="utf-8") as parsing_errors_log:
            gen_time = datetime.datetime.now(datetime.timezone.utc)
            parsing_errors_log.write(f"TibiaWikiSQL parsing errors - {gen_time.isoformat()}\n\n")
            parsing_errors_count = parse_articles(conn, data_store, enabled_categories, parsing_errors_log)
        click.echo(
            f"{Fore.YELLOW}Wrote {parsing_errors_count:,} parsing errors to "
            f"{parsing_errors_path}.{Style.RESET_ALL}",
        )
    else:
        parse_articles(conn, data_store, enabled_categories)

    for position in rashid_positions:
        RashidPositionTable.insert(conn, **position.model_dump())

    run_post_tasks(conn, data_store, enabled_categories, skip_images)

    with conn:
        gen_time = datetime.datetime.now(tz=datetime.timezone.utc)
        schema.DatabaseInfoTable.insert(conn, key="timestamp", value=str(gen_time.timestamp()))
        schema.DatabaseInfoTable.insert(conn, key="generate_time", value=gen_time.isoformat())
        schema.DatabaseInfoTable.insert(conn, key="version", value=__version__)
        schema.DatabaseInfoTable.insert(conn, key="python_version", value=platform.python_version())
        schema.DatabaseInfoTable.insert(conn, key="platform", value=platform.platform())