add `m` into the store, if it exists first remove it and its mappings each man page may have aliases besides the name determined by its basename
(self, m: ParsedManpage, raw: RawManpage)
| 472 | ) |
| 473 | |
| 474 | def add_manpage(self, m: ParsedManpage, raw: RawManpage) -> ParsedManpage: |
| 475 | """add `m` into the store, if it exists first remove it and its mappings |
| 476 | |
| 477 | each man page may have aliases besides the name determined by its |
| 478 | basename""" |
| 479 | validate_source_path(m.source) |
| 480 | existing = self._conn.execute( |
| 481 | "SELECT source FROM parsed_manpages WHERE source = ?", (m.source,) |
| 482 | ).fetchone() |
| 483 | if existing: |
| 484 | logger.debug("removing old manpage %s", m.source) |
| 485 | # Non-alias mappings (e.g. symlink-derived) will be lost to the |
| 486 | # CASCADE delete, but are recreated automatically by the |
| 487 | # symlink/dedup mapping phase at the end of the run. |
| 488 | alias_srcs = {a for a, _ in m.aliases} |
| 489 | lost = [ |
| 490 | row["src"] |
| 491 | for row in self._conn.execute( |
| 492 | "SELECT src FROM mappings WHERE dst = ?", (m.source,) |
| 493 | ).fetchall() |
| 494 | if row["src"] not in alias_srcs |
| 495 | ] |
| 496 | if lost: |
| 497 | logger.debug( |
| 498 | "re-importing %s will temporarily drop non-alias mappings: %s", |
| 499 | m.source, |
| 500 | ", ".join(sorted(lost)), |
| 501 | ) |
| 502 | # ON DELETE CASCADE removes all mappings rows for this manpage. |
| 503 | self._conn.execute( |
| 504 | "DELETE FROM parsed_manpages WHERE source = ?", (m.source,) |
| 505 | ) |
| 506 | self._conn.commit() |
| 507 | logger.debug("removed manpage and its mappings for %s", m.source) |
| 508 | else: |
| 509 | # Check for duplicate: same distro/release + name + section but different source |
| 510 | distro, release = config.parse_distro_release(m.source) |
| 511 | section = m.section |
| 512 | prefix = f"{distro}/{release}/" |
| 513 | conflict = self._conn.execute( |
| 514 | "SELECT source FROM parsed_manpages WHERE source LIKE ? AND source != ? AND name = ?", |
| 515 | (prefix + "%", m.source, m.name), |
| 516 | ).fetchone() |
| 517 | if conflict: |
| 518 | conflict_source = conflict["source"] |
| 519 | _, conflict_section = util.name_section( |
| 520 | os.path.basename(conflict_source)[:-3] |
| 521 | ) |
| 522 | if conflict_section == section: |
| 523 | raise errors.DuplicateManpage( |
| 524 | f"manpage {m.name}({section}) already exists with source " |
| 525 | f"{conflict_source!r}, refusing to add duplicate from {m.source!r}" |
| 526 | ) |
| 527 | |
| 528 | self._upsert_raw_manpage(m.source, raw) |
| 529 | |
| 530 | self._conn.execute( |
| 531 | """INSERT INTO parsed_manpages(source, name, synopsis, options, aliases, |