ex-ussr-kinship-research

83 files shipped with this skill. Inline text is the catalog copy. The upstream link is the source repository.

• SKILL.md54 KB9d2230295516truncatedupstream ↗

  Showing the inlined prefix. The rest is in the upstream file.

  ---
  name: ex-ussr-kinship-research
  description: Исследует родственные связи и семейную историю по персональным данным и документам с фокусом на бывший СССР, СССР и Российскую империю. Использовать для генеалогического поиска, сопоставления однофамильцев, архивных, метрических, военных, миграционных и репрессивных записей, построения доказательной цепочки и итогового отчёта. Не использовать для слежки за живыми людьми, поиска их текущих контактов или адресов, доступа к закрытым данным, обхода защиты и утверждения биологического родства без независимого подтверждения. EN — evidence-first genealogy and family-history research for the former USSR, the USSR and the Russian Empire (church/metric books, civil registration, military, migration, deportation and repression records, unindexed archive scans, namesake disambiguation, evidence chains, final report). Reads existing family trees as leads and exports source-cited trees in GEDCOM format when requested. Replies in the user's language. Not for tracking living people.
  compatibility: Предпочтительны внутренний браузер или веб-поиск, чтение PDF/DOCX/изображений и создание файлов. Использует доступные средства веб-поиска, каталоги/API и локальные материалы; отсутствие одного инструмента не останавливает остальные проверки. Работает как skill для Codex и Claude Code.
  metadata:
    version: "1.31.0"
    language: "ru"
    reply_language: "user"
    region: "ex-USSR"
  ---
  
  # Назначение
  
  Установить или опровергнуть родственные связи через воспроизводимое исследование. Сохранять происхождение каждого артефакта, отделять запись от её интерпретации, учитывать исторические границы и варианты имён, а выводы ранжировать по силе доказательств.
  
  # Язык общения / Language
  
  **Always reply in the user's language.** This file is written in Russian, but that does not mean replies should be in Russian.
  
  1. Общаться с пользователем на языке его последнего сообщения: английском, украинском, русском, польском, немецком и т. д. Если пользователь прямо просит другой язык — использовать его.
  2. Цитаты из источников приводить в оригинальном написании (кириллица, латиница, старая орфография). Рядом давать перевод на язык пользователя, если он нужен для понимания.
  3. Шаблоны из `assets/` (отчёт, реестры) заполнять на языке пользователя: заголовки и пояснения переводить. Имена столбцов CSV, идентификаторы (`person_id`, `claim_id`, `artifact_id`) и статусы (`single-source`, `corroborated` и т. д.) не переводить.
  4. Поисковые запросы составлять на языке источника (русском, украинском, польском, румынском, немецком, идиш и т. д.), а не на языке общения.
  5. Язык общения фиксировать в `case-intake.yaml` в поле `requested_output_language`. По умолчанию `auto` — язык пользователя.
  
  # Обновления навыка
  
  В начале исследования один раз проверить новые изменения в [каноническом репозитории](https://github.com/DenisSergeevitch/ex-ussr-kinship-research). Если есть обновление, уведомить пользователя и спросить: «Вышло обновление навыка. Обновить?» Агент пользователя выполнит обновление обычными доступными ему средствами после согласия. Пока ответа нет или пользователь отказался, продолжать исследование текущей версией без повторных предложений в каждом цикле.
  
  # Автономное исследование
  
  По умолчанию вести исследование самостоятельно в длительном, при необходимости многочасовом цикле. Не завершать работу после каждого этапа вопросом «продолжать?».
  
  1. Уточнения задавать один раз сообщением или неблокирующим инструментом, не переходя в ожидание ответа. Сразу работать с известными фактами; неизвестное отмечать как неизвестное, предположения — как гипотезы. Отсутствие ответа не блокирует исследование.
  2. Повторять цикл: выбрать следующую проверку из очереди → найти и прочитать источник → сохранить артефакт и происхождение → обновить утверждения и журнал → выбрать следующий полезный шаг. Проверенные поиски не повторять без новых оснований.
  3. CAPTCHA и вход обрабатывать по `references/browser-and-captcha.md`, разделы 5–5а. Если нужно участие пользователя, открыть соответствующую вкладку при наличии такой возможности и оставить шаг в очереди; блокируется только этот маршрут. Продолжать другие источники, варианты имени, семейные ветви и разбор уже полученных материалов.
  4. Сохранять находки в реестрах, а оставшиеся проверки и блокировки — в рабочем плане дела, например `research-state.md`. Использовать уже имеющийся план и не переписывать в него содержимое реестров. После перерыва продолжать с очередного полезного шага.
  5. Продолжать, пока остаются полезные доступные проверки в рамках задачи. Завершать при выполнении задачи и покрытии объёма, исчерпании доступных маршрутов, явной остановке пользователем или лимите среды. Перед вынужденным завершением сохранять состояние и достигнутый результат; один вопрос без ответа или один недоступный сайт не являются завершением всей работы.
  
  Если несколько исполнителей работают с общим делом или служебная обработка растёт быстрее полезного результата, читать `references/research-coordination.md`. Это условные рекомендации по разделению записи, передаче результатов и объёму обработки; они не требуют запускать дополнительных агентов или менять выбранное пользователем хранилище.
  
  # Генеалогическое древо в GEDCOM
  
  Если пользователь просит построить или экспортировать древо, подготовить локальный файл `.ged` по `references/gedcom.md`: GEDCOM 5.5.1 в UTF-8 по умолчанию для совместимости, либо версию, указанную пользователем или поддерживаемую выбранной им программой, например GEDCOM 7. Сохранить ссылки на источники и степень доказанности; реестры, журнал и нерешённые версии остаются в деле. Имеющееся древо разбирать как исходный материал по `references/source-strategy.md`, раздел 7, сохраняя оригинал.
  
  Программу для древа и место результата выбирает пользователь. Установка приложений, подключение MCP и автоматическая синхронизация внешнего древа находятся вне области навыка; наличие подключения само по себе не выбирает приложение и не поручает запись в него.
  
  # Работа с FamilySearch через MCP
  
  Если доступны инструменты FamilySearch MCP, использовать их для соответствующих операций в пределах задачи по `references/familysearch-mcp.md`: сначала проверить аккаунт, среду и фактические возможности подключения. Структурированное чтение, разрешённые изменения древа и просмотр неиндексированных сканов — разные маршруты; подключение MCP не заменяет оценку источников и не расширяет доступ. Без MCP продолжать обычное исследование через браузер и локальные материалы.
  
  # Неподлежащие нарушению правила
  
  1. Считать любой новый материал **артефактом, а не фактом**. Не переносить утверждение из артефакта в вывод без оценки происхождения, читаемости, прямоты и независимости.
  2. Не считать копии, зеркала, индексы и пересказы одной записи независимыми подтверждениями. Назначать им общий `independence_group`.
  3. Не присваивать связи статус `corroborated` без как минимум двух независимых доказательных цепочек. Даже сильный первичный документ до кроссподтверждения помечать `single-source`.
  4. Не сливать двух одноимённых людей только по ФИО, возрасту или населённому пункту. До разрешения личности вести отдельные `person_id`.
  5. Сохранять исходное написание имён, дат и мест. Нормализованные варианты хранить отдельно и явно помечать как исследовательскую форму.
  6. Не скрывать противоречия. Фиксировать их рядом с выводом и сохранять альтернативные гипотезы.
  7. Для живых людей минимизировать данные. Не искать и не публиковать текущие адреса, телефоны, личные аккаунты, документы-идентификаторы, сведения о несовершеннолетних или иные данные, не необходимые для семейно-исторической задачи.
  8. Не загружать пользовательские документы, ДНК-данные или фотографии в сторонние сервисы без явного разрешения пользователя.
  9. Не обходить ограничения доступа и лимиты. Для CAPTCHA допустима одна попытка через обычный интерфейс внутреннего браузера агента, если позволяют инструменты и правила сайта; вход выполняет пользователь. Порядок попытки, открытия вкладок и передачи 

• README.md14 KB1292d8799755upstream ↗

  # ex-USSR Kinship Research
  
  ![ex-USSR Kinship Research cover](docs/cover.jpg)
  
  Languages: [English](README.md) / [Русский](README.ru.md) / [Беларуская](README.be.md) / [Қазақша](README.kk.md) / [Українська](README.uk.md) / [Polski](README.pl.md) / [Deutsch](README.de.md) / [Română](README.ro.md) / [Eesti](README.et.md) / [Latviešu](README.lv.md) / [Lietuvių](README.lt.md) / [ქართული](README.ka.md) / [Հայերեն](README.hy.md) / [Azərbaycanca](README.az.md) / [Кыргызча](README.ky.md) / [Тоҷикӣ](README.tg.md) / [Türkmençe](README.tk.md) / [O‘zbekcha](README.uz.md) / [日本語](README.ja.md)
  
  **An evidence-first genealogy skill for AI agents (Codex, Claude Code and other Agent Skills runtimes).** It turns an agent into a careful family-history researcher for the former USSR, the Soviet Union and the Russian Empire. The agent reads church books and civil records, military, deportation and repression files, and unindexed archive scans. It keeps every find traceable, separates records from interpretation and says plainly what is proven and what is not.
  
  Coverage includes all 15 former Soviet republics: Russia, Ukraine, Belarus, Moldova, Estonia, Latvia, Lithuania, Georgia, Armenia, Azerbaijan, Kazakhstan, Kyrgyzstan, Tajikistan, Turkmenistan and Uzbekistan. Archives are chosen by the event's historical jurisdiction and current record custodian; a regional example does not replace local research.
  
  The skill's instructions are written in Russian, but **the agent talks to you in your language**: English, Ukrainian, Russian, Polish, German and so on. It quotes sources in their original spelling with a translation, and builds search queries in the languages of the records.
  
  ---
  
  ## What it does
  
  - **Case intake and scope.** One research question, known facts, who may be alive, what is off-limits.
  - **Reproducible registers.** Persons, artifacts, claims, relationships and a full search log (CSV), plus a final report template.
  - **Historical jurisdiction model.** It resolves every event to the state, province, district, parish or civil registry *as of the event date*, not today's map.
  - **Name-variant matrices.** Pre-reform Russian, Ukrainian, Polish, Romanian, German and Yiddish forms, patronymics, maiden names and transliterations.
  - **Source strategy.** Originals first, then official catalogs, indexes, published document collections, and only then trees and forums as leads.
  - **Evidence protocol.** Claims get explainable statuses, not percentages: `hypothesis`, `single-source`, `corroborated`, `strongly-corroborated`, `contradicted`, `unresolved`. Copies and mirrors of one record share an `independence_group`, so they never count as independent proof.
  - **Identity resolution.** Namesakes stay separate `person_id`s until distinguishing evidence appears: house number, parents, godparents, exact birth date or military unit.
  - **Audit script.** It checks referential integrity and flags weak corroboration before you write conclusions. An independence group counts only when a cited artifact belongs to it, so corroboration cannot be declared into existence. It also checks IDs and statuses in the working Markdown report, if present; the agent checks the delivered HTML against the registers. A summary shows what is still uncorroborated or marked illegible.
  - **Institutional requests.** Draft and, when authorized, send focused requests to archives, libraries, universities and museums using available email or official web tools. If needed, offer a supported mail integration or a ready-to-send letter. Record the outcome, avoid duplicate sends and treat replies as evidence. See [correspondence guidance](references/correspondence.md).
  - **Archive request form.** An anonymized [CANO vital-record request template](assets/cano-vital-record-request.template.docx) preserves the supplied form's layout for an editable case copy. Verify the archive's current form and submission requirements before use.
  - **Government services.** Choose an official civil-registration or archive request route, checking eligibility and how the result is delivered ([government portals](references/government-portals.md)).
  - **Final report.** A complete HTML report is the default deliverable, with each relationship laid out edge by edge, contradictions, failed searches and concrete next steps. PDF is an optional export only when you explicitly request it.
  - **Targeted indexed-record research.** Find relevant FamilySearch books by place or DGS through an existing lawful browser session, view specific records, and retain only evidence fragments permitted by the source. Local helpers search relatives, witnesses and name variants in saved fragments, preserving their context and file checksums. [Workflow and coverage limits](references/familysearch-indexes.md).
  - **Connected FamilySearch MCP.** Use a connected server for structured searches, profiles, relatives and sources, and authorized tree changes with explicit scope and read-back. Keep indexed search separate from unindexed scan reading, and check the connector's actual browser, privacy and write capabilities. [Workflow and access limits](references/familysearch-mcp.md).
  - **WWI source search.** Search GWAR reports, hospital, award and prisoner-of-war records by person or locality, compare settlement spellings across saved cards to decode a distorted place name, then locate the source image. [Workflow](references/ww1-search.md). Both helpers feed the existing artifact register and search log; they do not establish identities or kinship automatically.
  
  Release history: [changelog](references/changelog.md).
  
  ## Install
  
  With Node.js installed, you can use the optional [Skills CLI](https://github.com/vercel-labs/skills):
  
  ```bash
  npx skills add https://github.com/DenisSergeevitch/ex-ussr-kinship-research
  ```
  
  The CLI lets you choose your agent and installs into the current project by default; add `-g` for a personal installation across projects. See its documentation for current Node.js requirements.
  
  Alternatively, clone the repository directly into your agent's skills folder:
  
  ```bash
  # Codex
  git clone https://github.com/DenisSergeevitch/ex-ussr-kinship-research ~/.codex/skills/ex-ussr-kinship-research
  
  # Claude Code (personal skills)
  git clone https://github.com/DenisSergeevitch/ex-ussr-kinship-research ~/.claude/skills/ex-ussr-kinship-research
  ```
  
  For a single project, put the folder in `.claude/skills/` inside the repository. Most scripts need only Python 3 and the standard library. The optional `scan_montage.py` helper needs Pillow; `build_pdf.py` needs a Chromium-based browser.
  
  ## Use
  
  Describe the family question. The agent works from available information and continues while questions or access requests await a reply. For example:
  
  - *"My great-grandfather was exiled to Kazakhstan in the 1930s. Help me find his parents in church records."*
  - *"Here are scans of a marriage certificate and a soldier card. Check whether they are the same person and build the evidence chain."*
  - *"Find the parish and the right FamilySearch film for a village in Podolia in the 1890s and read the births for the surname X."*
  
  Keep private case data outside the skill's Git checkout. From the skill folder:
  
  ```bash
  python scripts/init_case.py --case-dir ~/kinship-cases/my-case --research-question "Who were Maria's parents?" --subject-label "Maria"
  python scripts/hash_artifact.py ~/kinship-cases/my-case/scans/marriage-1902.jpg   # SHA-256 for every saved artifact
  python scripts/audit_evidence.py --case-dir ~/kinship-cases/my-case             # check registers before the final report
  ```
  
  `init_case.py` refuses to create a case inside the skill folder, including with
  `--force`. Git ignores common legacy case-folder names as an additional safeguard;
  check staged files before contributing, especially if moving an existing case.
  
  The agent prepares a complete HTML report, for example `my-case/report.html`, using `assets/report.template.html` or its existing tools. The default editorial layout puts a family tree in the header, with relationship evidence one click away, readable multilingual system-font fallbacks, and layouts for small screens and print. The tree follows the available evidence; missing relatives do not delay the report. Open it directly in a browser or share the HTML with its local images. Short answers may remain text or Markdown; an explicitly requested format takes precedence. A family-facing report does not imply PDF.
  
  Only when you explicitly request PDF, the agent can export the same HTML:
  
  ```bash
  python scripts/build_pdf.py --html ~/kinship-cases/my-case/report.html --out ~/kinship-cases/my-case/report.pdf
  ```
  
  This optional export needs Chrome, Chromium or Edge installed (or `CHROME_PATH`). With `--png-dir` it also renders pages to PNG for review if `pypdfium2` is available. HTML delivery needs no PDF renderer.
  
  For PDF export, use static HTML with local or embedded images. The exporter rejects active HTML and network image sources; source links remain clickable. A failed render preserves any existing output PDF.
  
  ## Family trees in GEDCOM
  
  [GEDCOM](https://gedcom.io/specs/) is a standard format for exchanging genealogical data between programs. When you ask the agent to build or export a tree, it prepares a local `.ged` file with source citations and evidence notes. The default is GEDCOM 5.5.1 in UTF-8 for compatibility; use another supported version, such as GEDCOM 7, when you or your chosen program require it.
  
  An existing tree is research input and a set of leads, not proof of its relationships. The agent preserves the original file and keeps unresolved claims, unsupported details, evidence registers and the search log in the case folder. Exports follow the evaluated evidence and do not overwrite your files.
  
  You choose the software. Installing tree applications, configuring MCP connections and automatically synchronizing external trees are outside the skill's scope. See [the GEDCOM workflow](references/gedcom.md).
  
  ## Repository layout
  
  ```
  SKILL.md                         main instructions (workflow, hard rules, language policy)
  references/
    source-strategy.md             source hierarchy, query patterns, catalogs, archive requests
    correspondence.md              recipients, mail integration, authorized sending and replies
    government-portals.md          civil-registration services and request delivery routes
    evidence-protocol.md           artifact assessment, statuses, identity resolution, conflicts
    ex-ussr-locale-guide.md        jurisdictions, name variants, calendars, record types by period
    browser-and-captcha.md         browsing, sign-in, one CAPTCHA attempt and hand-off, rate limits
    privacy-and-safety.md          living people, minors, DNA, sensitive findings
    seed-sources.md                curated starting resources with reliability classes
    field-lessons.md               practical lessons from a real case
    changelog.md                   release history
    photo-search.md                finding, identifying, dating and storing photos
    habsburg-galicia-bukovina.md   Galicia and Bukovina under the Habsburgs and after
    chechnya-deportation.md        Chechen families: 1944 deportation, special settlement, names and renamed villages
    germany-forced-labour.md       Forced labourers in Germany 1941-1945: Arolsen Archives and German civil registers
  
    bessarabia-moldova.md          Bessarabia, Moldova, MASSR and the Dniester borderland
    familysearch-indexes.md       targeted browser research, permitted local snapshots and coverage
    familysearch-mcp.md           connected tools, structured reads, scoped writes and verification
    ww1-search.md                 GWAR WWI person/locality search and source-image locators
    gedcom.md                     portable family-tree exports with sources and evidence notes
  assets/                          case intake, CSV registers, final report template,
                                   photos README template, research-state template, HTML report template (report.template.html)
  scripts/                         init_case.py, hash_artifact.py, audit_evidence.py, build_pdf.py
                                   familysearch.py, gwar.py, register_search.py, scan_montage.py (needs Pillow)
                                   and shared local analysis modules
  evals/discovery-and-logic.yaml   trigger and reasoning test cases
  tests/                           unit tests (python3 -m unittest discover -s tests)
  ```
  
  ## Responsible use
  
  - **Family history, not surveillance.** The skill refuses to look up current addresses, phones or accounts of living people, and it minimizes their data in reports.
  - **Access checks.** The agent may try an actual CAPTCHA once through normal controls in its internal browser when supported and allowed. If user action is needed, it opens the relevant tab when possible and provides the URL and required action; sign-ins are done by the user. No external solvers, protection bypass, paywall, geo-block or rate-limit evasion, or leaked databases. See [browser guidance](references/browser-and-captcha.md).
  - **No uploads of your documents or DNA data** to third-party services without your explicit consent.
  - **Sensitive findings** (deportation, repression, adoption, DNA surprises) are reported neutrally and only as far as the question requires.
  - **Authorized correspondence.** The skill can send institutional requests through available tools when the user has authorized them; it does not ask again for an already authorized send. Connecting a mailbox alone does not authorize sending, attachments or paid services. Verify the current official contact and procedure, and keep credentials and private case correspondence outside the skill repository.
  
  ## Contributing
  
  New sources are welcome in `references/seed-sources.md`. Record verified owner, jurisdiction, collection, provenance, IDs, access, restrictions and check date where known; mark gaps instead of guessing. Keep personal data out of examples.
  
  Source-search helpers use Python's standard library and the existing case registers. Run `python3 -m unittest discover -s tests` to check local parsing, GWAR checkpoints, access failures, unresolved coverage and evidence registration against synthetic data. FamilySearch discovery and targeted viewing use the existing lawful browser session; its Python helpers analyze saved fragments. Keep private research files outside this checkout.
  
  ## License
  
  [MIT](LICENSE) © 2026 Denis Shiryaev.
  
  ---
  
  Maintained by Denis Shiryaev ([@DenisSergeevitch](https://github.com/DenisSergeevitch)).
  

• scripts/audit_evidence.py21 KBca9f4e1f5f50upstream ↗

  #!/usr/bin/env python3
  """Audit cross-references and corroboration claims without assigning truth."""
  
  from __future__ import annotations
  
  import argparse
  import csv
  import json
  import re
  import sys
  from collections import Counter
  from pathlib import Path
  from typing import Iterable
  
  REQUIRED_FILES = {
      "artifact-register.csv": {
          "artifact_id",
          "independence_group",
          "directness",
          "originality",
          "related_claim_ids",
          "status",
      },
      "claims-register.csv": {
          "claim_id",
          "artifact_ids",
          "supporting_independence_groups",
          "contradicting_artifact_ids",
          "status",
          "rationale",
      },
      "relationship-register.csv": {
          "relationship_id",
          "person_a_id",
          "person_b_id",
          "claim_ids",
          "supporting_artifact_ids",
          "contradicting_artifact_ids",
          "independence_groups",
          "status",
          "rationale",
      },
      "person-register.csv": {"person_id"},
  }
  
  CORROBORATED = {"corroborated", "strongly-corroborated"}
  VALID_CLAIM_STATUSES = {
      "",
      "hypothesis",
      "single-source",
      "corroborated",
      "strongly-corroborated",
      "contradicted",
      "unresolved",
  }
  
  
  def split_ids(value: str) -> list[str]:
      if not value:
          return []
      normalized = value.replace(",", ";")
      return [part.strip() for part in normalized.split(";") if part.strip()]
  
  
  def read_csv(path: Path, required: set[str]) -> tuple[list[dict[str, str]], list[str]]:
      errors: list[str] = []
      try:
          with path.open("r", encoding="utf-8-sig", newline="") as handle:
              reader = csv.DictReader(handle)
              if reader.fieldnames is None:
                  return [], [f"{path.name}: отсутствует строка заголовка"]
              missing = sorted(required - set(reader.fieldnames))
              if missing:
                  errors.append(f"{path.name}: отсутствуют колонки: {', '.join(missing)}")
              rows = []
              for row in reader:
                  if None in row:
                      # csv.DictReader складывает лишние поля в ключ None: обычно незакавыченная запятая.
                      errors.append(f"{path.name}: строка {reader.line_num} содержит больше полей, чем заголовок (незакавыченная запятая?)")
                      row.pop(None)
                  if any(value is None for value in row.values()):
                      # Недостающие поля DictReader заполняет None: значения сдвинулись относительно заголовка.
                      present = sum(value is not None for value in row.values())
                      errors.append(
                          f"{path.name}: строка {reader.line_num} содержит меньше полей, чем заголовок "
                          f"({present} из {len(reader.fieldnames)}): значения могли сдвинуться в чужие колонки"
                      )
                  rows.append({k: (v or "").strip() for k, v in row.items()})
              return rows, errors
      except OSError as exc:
          return [], [f"{path.name}: ошибка чтения: {exc}"]
  
  
  def duplicate_ids(rows: Iterable[dict[str, str]], field: str) -> list[str]:
      ids = [row.get(field, "") for row in rows if row.get(field, "")]
      return sorted(key for key, count in Counter(ids).items() if count > 1)
  
  
  def artifact_groups(artifact_ids: Iterable[str], artifact_by_id: dict[str, dict[str, str]]) -> set[str]:
      return {
          artifact_by_id[artifact_id]["independence_group"].strip().casefold()
          for artifact_id in artifact_ids
          if artifact_id in artifact_by_id
          and artifact_by_id[artifact_id].get("independence_group", "").strip()
      }
  
  
  def has_direct_record(artifact_ids: Iterable[str], artifact_by_id: dict[str, dict[str, str]]) -> bool:
      """Одна и та же запись должна быть прямой и иметь описанное происхождение.
  
      Непустое описание и локатор — проверка заполнения, не доказательство
      подлинности документа или истинности заключения.
      """
      for artifact_id in artifact_ids:
          row = artifact_by_id.get(artifact_id, {})
          provenance = row.get("provenance_level", "").strip().casefold()
          if (
              row.get("directness", "").casefold() == "direct"
              and row.get("originality", "").casefold()
              in {"original", "image-of-original", "official-extract"}
              and provenance not in {"", "unknown", "неизвестно", "не установлено"}
              and (row.get("source_url_or_path", "").strip() or row.get("archival_reference", "").strip())
          ):
              return True
      return False
  
  
  def id_prefix(value: str) -> str:
      """Префикс ID без номера: 'A-0001' -> 'A-'."""
      return value.rstrip("0123456789")
  
  
  REPORT_MARKERS = ("[неразборчиво]", "[предположительно")
  HEADING_RE = re.compile(r"^#{2,6}\s")
  
  
  def audit_report(
      text: str,
      known_ids: dict[str, set[str]],
      statuses: dict[str, str],
  ) -> tuple[list[str], dict[str, int]]:
      """Сверить ID и статусы итогового отчёта с реестрами.
  
      known_ids: префикс -> ID этого префикса из реестров. Стандартные префиксы
      проверяются и при пустых реестрах; плейсхолдеры (`A-....`) не содержат номера.
      statuses: ID утверждения или связи -> статус в реестре.
      """
      errors: list[str] = []
      for prefix, ids in sorted(known_ids.items()):
          pattern = re.compile(rf"(?<![\w-]){re.escape(prefix)}\d+(?!\w)")
          for cited in sorted(set(pattern.findall(text)) - ids):
              errors.append(f"final-report.md: ссылка на {cited}, которого нет в реестрах")
  
      valid = VALID_CLAIM_STATUSES - {""}
      section_id = ""
      section_checked = False
      for number, line in enumerate(text.splitlines(), start=1):
          stripped = line.strip()
          if HEADING_RE.match(stripped):
              words = re.findall(r"[\w-]+", stripped)
              section_id = next((word for word in words if word in statuses), "")
              section_checked = False
              continue
          if stripped.startswith("|"):
              # Строка таблицы с ID в первой ячейке и статусом в одной из ячеек.
              cells = [cell.strip().strip("`") for cell in stripped.strip("|").split("|")]
              row_id, row_statuses = cells[0], [cell for cell in cells[1:] if cell in valid]
              if row_id in statuses and row_statuses and statuses[row_id] not in row_statuses:
                  errors.append(
                      f"final-report.md:{number}: {row_id} — в отчёте {row_statuses[0]}, в реестре {statuses[row_id] or 'пусто'}"
                  )
              continue
          # Первое поле вида **Статус:** value в секции связи/утверждения.
          match = re.match(r"^\*\*[^*]+:\*\*\s*`?([\w-]+)`?", stripped)
          if section_id and not section_checked and match and match.group(1) in valid:
              section_checked = True
              if match.group(1) != statuses[section_id]:
                  errors.append(
                      f"final-report.md:{number}: {section_id} — в отчёте {match.group(1)}, в реестре {statuses[section_id] or 'пусто'}"
                  )
  
      markers = {marker.strip("[]"): text.count(marker) for marker in REPORT_MARKERS}
      return errors, markers
  
  
  def parse_args() -> argparse.Namespace:
      parser = argparse.ArgumentParser(
          description="Проверяет целостность реестров и минимальные условия кроссподтверждения."
      )
      parser.add_argument("--case-dir", required=True, help="Директория дела")
      parser.add_argument("--json", action="store_true", help="Вывести JSON")
      parser.add_argument(
          "--strict",
          action="store_true",
          help="Возвращать ненулевой код также при предупреждениях",
      )
      return parser.parse_args()
  
  
  def main() -> int:
      args = parse_args()
      case_dir = Path(args.case_dir).expanduser().resolve()
      errors: list[str] = []
      warnings: list[str] = []
  
      if not case_dir.is_dir():
          print(f"Ошибка: директория дела не найдена: {case_dir}", file=sys.stderr)
          return 2
  
      data: dict[str, list[dict[str, str]]] = {}
      for filename, required in REQUIRED_FILES.items():
          path = case_dir / filename
          if not path.is_file():
              errors.append(f"Отсутствует обязательный файл: {filename}")
              data[filename] = []
              continue
          rows, file_errors = read_csv(path, required)
          errors.extend(file_errors)
          data[filename] = rows
  
      artifacts = data["artifact-register.csv"]
      claims = data["claims-register.csv"]
      relationships = data["relationship-register.csv"]
      persons = data["person-register.csv"]
  
      for rows, field, label in [
          (artifacts, "artifact_id", "артефакта"),
          (claims, "claim_id", "утверждения"),
          (relationships, "relationship_id", "связи"),
          (persons, "person_id", "лица"),
      ]:
          duplicates = duplicate_ids(rows, field)
          if duplicates:
              errors.append(f"Дублирующиеся ID {label}: {', '.join(duplicates)}")
          for index, row in enumerate(rows, start=2):
              if not row.get(field, ""):
                  warnings.append(f"{field}: пустой ID в строке {index}")
  
      artifact_by_id = {row["artifact_id"]: row for row in artifacts if row.get("artifact_id")}
      claim_by_id = {row["claim_id"]: row for row in claims if row.get("claim_id")}
      person_ids = {row["person_id"] for row in persons if row.get("person_id")}
  
      for artifact_id, row in artifact_by_id.items():
          if not row.get("independence_group", "").strip():
              warnings.append(f"{artifact_id}: не указан independence_group")
          for person_id in split_ids(row.get("person_ids", "")):
              if person_id not in person_ids:
                  errors.append(f"{artifact_id}: неизвестный person_id {person_id}")
          for claim_id in split_ids(row.get("related_claim_ids", "")):
              if claim_id not in claim_by_id:
                  errors.append(f"{artifact_id}: неизвестный related claim_id {claim_id}")
  
      for claim_id, row in claim_by_id.items():
          status = row.get("status", "")
          if status not in VALID_CLAIM_STATUSES:
              warnings.append(f"{claim_id}: неизвестный статус {status!r}")
          for person_id in split_ids(row.get("subject_person_ids", "")):
              if person_id not in person_ids:
                  errors.append(f"{claim_id}: неизвестный subject_person_id {person_id}")
          referenced = split_ids(row.get("artifact_ids", ""))
          contradictory = split_ids(row.get("contradicting_artifact_ids", ""))
          for artifact_id in referenced + contradictory:
              if artifact_id not in artifact_by_id:
                  errors.append(f"{claim_id}: неизвестный artifact_id {artifact_id}")
  
          groups = artifact_groups(referenced, artifact_by_id)
          for group in sorted(set(split_ids(row.get("supporting_independence_groups", "")))):
              if group.casefold() not in groups:
                  errors.append(
                      f"{claim_id}: independence_group {group} заявлен, но ни один из artifact_ids к нему не относится"
                  )
  
          if status in CORROBORATED and len(groups) < 2:
              warnings.append(
                  f"{claim_id}: статус {status}, но найдено независимых групп: {len(groups)}"
              )
          if status in CORROBORATED and not row.get("rationale"):
              warnings.append(f"{claim_id}: для {status} отсутствует rationale")
          if status in CORROBORATED and contradictory:
              warnings.append(
                  f"{claim_id}: статус {status}, но указаны противоречащие артефакты; "
                  "проверьте и объясните, разрешён ли существенный конфликт"
              )
          if status == "single-source" and len(groups) > 1:
              warnings.append(
                  f"{claim_id}: несколько независимых групп, но статус всё ещё single-source; проверьте конфликт/полноту"
              )
          if status == "strongly-corroborated" and not has_direct_record(referenced, artifact_by_id):
              warnings.append(
                  f"{claim_id}: strongly-corroborated без единой прямой записи "
                  "с указанным происхождением и локатором оригинала, образа или официальной выписки"
              )
  
      for row in relationships:
          relation_id = row.get("relationship_id", "<без ID>")
          status = row.get("status", "")
          if status not in VALID_CLAIM_STATUSES:
              warnings.append(f"{relation_id}: неизвестный статус {status!r}")
          for person_field in ("person_a_id", "person_b_id"):
              person_id = row.get(person_field, "")
              if person_id and person_id not in person_ids:
                  errors.append(f"{relation_id}: неизвестный {person_field} {person_id}")
          for claim_id in split_ids(row.get("claim_ids", "")):
              if claim_id not in claim_by_id:
                  errors.append(f"{relation_id}: неизвестный claim_id {claim_id}")
          artifact_refs = split_ids(row.get("supporting_artifact_ids", "")) + split_ids(
              row.get("contradicting_artifact_ids", "")
          )
          for artifact_id in artifact_refs:
              if artifact_id not in artifact_by_id:
                  errors.append(f"{relation_id}: неизвестный artifact_id {artifact_id}")
          # Группа засчитывается, только если за ней стоит артефакт: прямой или через claim_ids.
          backing_ids = split_ids(row.get("supporting_artifact_ids", ""))
          contradictory = set(split_ids(row.get("contradicting_artifact_ids", "")))
          for claim_id in split_ids(row.get("claim_ids", "")):
              if claim_id in claim_by_id:
                  backing_ids += split_ids(claim_by_id[claim_id].get("artifact_ids", ""))
                  contradictory.update(split_ids(claim_by_id[claim_id].get("contradicting_artifact_ids", "")))
          groups = artifact_groups(backing_ids, artifact_by_id)
          for group in sorted(set(split_ids(row.get("independence_groups", "")))):
              if group.casefold() not in groups:
                  errors.append(
                      f"{relation_id}: independence_group {group} заявлен, но ни один из артефактов связи к нему не относится"
                  )
          if status in CORROBORATED:
              if len(groups) < 2:
                  warnings.append(
                      f"{relation_id}: статус {row.get('status')}, но указано независимых групп: {len(groups)}"
                  )
              if not row.get("rationale"):
                  warnings.append(f"{relation_id}: отсутствует rationale")
              if contradictory:
                  warnings.append(
                      f"{relation_id}: статус {status}, но указаны противоречащие артефакты; "
                      "проверьте и объясните, разрешён ли существенный конфликт"
                  )
          if status == "strongly-corroborated" and not has_direct_record(backing_ids, artifact_by_id):
              warnings.append(
                  f"{relation_id}: strongly-corroborated без единой прямой записи "
                  "с указанным происхождением и локатором оригинала, образа или официальной выписки"
              )
  
      # Итоговый отчёт: ссылки и статусы должны совпадать с реестрами.
      markers: dict[str, int] = {}
      report = case_dir / "final-report.md"
      if report.is_file():
          known_ids: dict[str, set[str]] = {
              prefix: set() for prefix in ("A-", "C-", "R-", "P-")
          }
          for rows, field in [
              (artifacts, "artifact_id"),
              (claims, "claim_id"),
              (relationships, "relationship_id"),
              (persons, "person_id"),
          ]:
              for row in rows:
                  if row.get(field, "") and id_prefix(row[field]) != row[field]:
                      known_ids.setdefault(id_prefix(row[field]), set()).add(row[field])
          statuses = {claim_id: row.get("status", "") for claim_id, row in claim_by_id.items()}
          statuses.update(
              (row["relationship_id"], row.get("status", ""))
              for row in relationships
              if row.get("relationship_id")
          )
          report_text = report.read_text(encoding="utf-8-sig")
          report_errors, markers = audit_report(report_text, known_ids, statuses)
          errors.extend(report_errors)
  
      # Журнал поисков необязателен; находка без артефакта — предупреждение.
      search_rows: list[dict[str, str]] = []
      search_log = case_dir / "search-log.csv"
      if search_log.is_file():
          search_rows, search_errors = read_csv(search_log, {"search_id", "result_status", "artifact_ids"})
          warnings.extend(search_errors)
          for row in search_rows:
              for artifact_id in split_ids(row.get("artifact_ids", "")):
                  if artifact_id not in artifact_by_id:
                      warnings.append(
                          f"{row.get('search_id') or '<без ID>'}: неизвестный artifact_id {artifact_id} в журнале поиска"
                      )
              if row.get("result_status", "").strip().casefold() == "found" and not row.get(
                  "artifact_ids", ""
              ).strip():
                  warnings.append(
                      f"{row.get('search_id') or '<без ID>'}: result_status=found, но artifact_ids пуст — "
                      "находка не зарегистрирована как артефакт"
                  )
  
      # Сводка готовности: что ещё не снято. Не ошибки — для решения, выдавать ли отчёт.
      summary = {
          "claims_by_status": dict(Counter(row.get("status", "") or "пусто" for row in claims)),
          "relationships_by_status": dict(
              Counter(row.get("status", "") or "пусто" for row in relationships)
          ),
          "report_markers": markers,
      }
  
      result = {
          "case_dir": str(case_dir),
          "counts": {
              "persons": len(persons),
              "artifacts": len(artifacts),
              "claims": len(claims),
              "relationships": len(relationships),
              "searches": len(search_rows),
          },
          "summary": summary,
          "errors": errors,
          "warnings": warnings,
      }
  
      if args.json:
          print(json.dumps(result, ensure_ascii=False, indent=2))
      else:
          print(f"Проверено дело: {case_dir}")
          print(
              "Записей: "
              f"лица={len(persons)}, артефакты={len(artifacts)}, "
              f"утверждения={len(claims)}, связи={len(relationships)}, поиски={len(search_rows)}"
          )
          open_statuses = ("hypothesis", "single-source", "unresolved", "contradicted", "пусто")
          print("\nСВОДКА:")
          for label, key in [("утверждения", "claims_by_status"), ("связи", "relationships_by_status")]:
              counts = summary[key]
              line = ", ".join(f"{status}={counts[status]}" for status in sorted(counts))
              pending = sum(counts.get(status, 0) for status in open_statuses)
              print(f"- {label}: {line or 'нет'}; без кроссподтверждения: {pending}")
          if markers:
              print("- в отчёте: " + ", ".join(f"[{name}]={count}" for name, count in markers.items()))
          if errors:
              print("\nОШИБКИ:")
              for item in errors:
                  print(f"- {item}")
          if warnings:
              print("\nПРЕДУПРЕЖДЕНИЯ:")
              for item in warnings:
                  print(f"- {item}")
          if not errors and not warnings:
              print("Ошибок и предупреждений не найдено.")
  
      if errors:
          return 1
      if warnings and args.strict:
          return 1
      return 0
  
  
  if __name__ == "__main__":
      sys.path.insert(0, str(Path(__file__).resolve().parent))
      from research_common import configure_utf8_stdio
  
      configure_utf8_stdio()
      raise SystemExit(main())
  

• scripts/build_pdf.py13 KB450279ed15d7upstream ↗

  #!/usr/bin/env python3
  """Build the human-readable PDF report from an HTML file via headless Chrome/Chromium."""
  
  from __future__ import annotations
  
  import argparse
  import html
  import html.parser
  import os
  import re
  import shutil
  import subprocess
  import sys
  import tempfile
  from pathlib import Path
  from urllib.parse import unquote, urlsplit
  from urllib.request import url2pathname
  
  PLACEHOLDER_RE = re.compile(r"<[^<>\n]+>")
  
  CHROME_NAMES = [
      "google-chrome",
      "google-chrome-stable",
      "chromium",
      "chromium-browser",
      "chrome",
      "msedge",
      "microsoft-edge",
  ]
  CHROME_PATHS = [
      "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
      "/Applications/Chromium.app/Contents/MacOS/Chromium",
      "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
      r"C:\Program Files\Google\Chrome\Application\chrome.exe",
      r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
      r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
  ]
  
  STATIC_TAGS = set(
      "html head body title meta style a abbr address area article aside b bdi bdo "
      "blockquote br caption cite code col colgroup data dd del details dfn div dl dt "
      "em figcaption figure footer h1 h2 h3 h4 h5 h6 header hgroup hr i img ins kbd "
      "li main map mark menu nav ol p pre q rp rt ruby s samp section small span "
      "strong sub summary sup table tbody td th thead time tr u ul var wbr".split()
  )
  ACTIVE_TAGS = {
      "script", "iframe", "frame", "frameset", "object", "embed", "applet",
      "portal", "fencedframe", "form", "input", "button", "select", "textarea",
      "base", "link",
  }
  PRINT_CSP = (
      "default-src 'none'; script-src 'none'; connect-src 'none'; "
      "frame-src 'none'; object-src 'none'; media-src 'none'; "
      "img-src file: data:; font-src file: data:; style-src 'unsafe-inline'; "
      "base-uri file:; form-action 'none'"
  )
  
  
  class VisibleText(html.parser.HTMLParser):
      """Collect text and images, rejecting active HTML and unknown placeholder tags."""
  
      def __init__(self) -> None:
          super().__init__(convert_charrefs=True)
          self.skip = 0
          self.chunks: list[str] = []
          self.images: list[str] = []
          self.problems: list[str] = []
  
      def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
          if tag in ACTIVE_TAGS:
              self.problems.append(f"активный или внешний HTML запрещён: <{tag}>")
          elif tag not in STATIC_TAGS:
              self.problems.append(
                  "незаполненный плейсхолдер или неподдерживаемый тег: "
                  + (self.get_starttag_text() or f"<{tag}>")
              )
          attributes = dict(attrs)
          if tag == "meta" and (attributes.get("http-equiv") or "").strip().lower() == "refresh":
              self.problems.append("автоматический переход meta refresh запрещён")
          for key, value in attrs:
              if key.startswith("on") or key in {"srcdoc", "ping", "action", "formaction", "srcset"}:
                  self.problems.append(f"активный или неподдерживаемый атрибут запрещён: {key}")
              if value and PLACEHOLDER_RE.search(value):
                  self.problems.append(f"незаполненный плейсхолдер в атрибуте {key}: {value}")
              if key == "href" and tag in {"a", "area"} and value:
                  try:
                      target = urlsplit(re.sub(r"[\x00-\x20\x7f]", "", value))
                  except ValueError:
                      self.problems.append(f"некорректная ссылка: {value}")
                      continue
                  if target.scheme.lower() not in {"", "http", "https", "mailto", "file"}:
                      self.problems.append(f"небезопасная схема ссылки: {value}")
                  elif (not target.scheme or target.scheme == "file") and target.netloc not in {"", "localhost"}:
                      self.problems.append(f"сетевая файловая ссылка запрещена: {value}")
          if tag in ("style", "script", "title"):
              self.skip += 1
          if tag == "img":
              src = dict(attrs).get("src") or ""
              self.images.append(src)
  
      def handle_endtag(self, tag: str) -> None:
          if tag in ("style", "script", "title") and self.skip:
              self.skip -= 1
  
      def handle_data(self, data: str) -> None:
          if not self.skip:
              self.chunks.append(data)
  
      def unknown_decl(self, data: str) -> None:  # pragma: no cover - defensive
          pass
  
  
  def parse_args() -> argparse.Namespace:
      parser = argparse.ArgumentParser(
          description="Собирает читаемую PDF-версию отчёта из HTML (SKILL.md, шаг 12)."
      )
      parser.add_argument("--html", required=True, help="HTML-файл отчёта (обычно <дело>/report.html)")
      parser.add_argument("--out", help="Путь к PDF; по умолчанию рядом с HTML")
      parser.add_argument(
          "--check-only",
          action="store_true",
          help="Только проверить текст и изображения, PDF не собирать",
      )
      # Accepted for older invocations; presentation choices no longer block printing.
      parser.add_argument("--allow-service-ids", action="store_true", help=argparse.SUPPRESS)
      parser.add_argument("--chrome", help="Путь к Chrome/Chromium/Edge; иначе переменная CHROME_PATH или автопоиск")
      parser.add_argument(
          "--png-dir",
          help="Отрисовать страницы PDF в PNG для визуальной проверки (нужен установленный pypdfium2)",
      )
      return parser.parse_args()
  
  
  def inspect_html(source: str, html_path: Path) -> tuple[list[str], list[str]]:
      parser = VisibleText()
      parser.feed(source)
      parser.close()
      text = re.sub(r"\s+", " ", "".join(parser.chunks))
  
      problems = parser.problems[:]
      placeholders = sorted(set(PLACEHOLDER_RE.findall(text)))
      if placeholders:
          problems.append("незаполненный плейсхолдер <…>: " + ", ".join(placeholders[:10]))
  
      missing: list[str] = []
      for src in parser.images:
          if not src or "<" in src:
              missing.append(src or "<пустой src>")
              continue
          try:
              target = urlsplit(src)
          except ValueError:
              problems.append(f"некорректный путь изображения: {src}")
              continue
          if target.scheme.lower() == "data" and src.lower().startswith("data:image/"):
              continue
          if target.scheme.lower() not in {"", "file"} or target.netloc not in {"", "localhost"}:
              problems.append(f"изображение должно быть локальным или data:image: {src}")
              continue
          image_path = Path(url2pathname(target.path)) if target.scheme else Path(unquote(target.path))
          if not (html_path.parent / image_path).is_file():
              missing.append(src)
      return problems, missing
  
  
  def check_html(html_path: Path) -> tuple[list[str], list[str]]:
      return inspect_html(html_path.read_text(encoding="utf-8"), html_path)
  
  
  def find_chrome(explicit: str | None) -> str | None:
      for candidate in (explicit, os.environ.get("CHROME_PATH")):
          if candidate and Path(candidate).exists():
              return candidate
      for name in CHROME_NAMES:
          found = shutil.which(name)
          if found:
              return found
      for path in CHROME_PATHS:
          if Path(path).exists():
              return path
      return None
  
  
  def print_pdf(chrome: str, html_path: Path, out_path: Path) -> int:
      html_path, out_path = html_path.resolve(), out_path.resolve()
      if html_path == out_path or (out_path.exists() and html_path.samefile(out_path)):
          print("Ошибка: исходный HTML и выходной PDF должны быть разными файлами.", file=sys.stderr)
          return 2
      source = html_path.read_text(encoding="utf-8")
      problems, missing = inspect_html(source, html_path)
      if problems or missing:
          print("Ошибка: HTML не прошёл проверку перед печатью: " + "; ".join(problems + missing), file=sys.stderr)
          return 1
      # The first policy is enforced before any report content is parsed. A trusted
      # local base preserves relative image paths while the temporary copy is printed.
      printable = (
          '<!doctype html>\n<meta charset="utf-8">\n'
          f'<meta http-equiv="Content-Security-Policy" content="{html.escape(PRINT_CSP, quote=True)}">\n'
          f'<base href="{html.escape(html_path.parent.as_uri() + "/", quote=True)}">\n'
          + source
      )
      diagnostics = ""
      with tempfile.TemporaryDirectory(prefix=".kinship-pdf-", dir=out_path.parent) as staging:
          stage = Path(staging)
          staged_html, staged_pdf = stage / "report.html", stage / "report.pdf"
          staged_html.write_text(printable, encoding="utf-8")
          base = [
              chrome,
              "--disable-gpu",
              "--disable-background-networking",
              "--disable-extensions",
              "--disable-sync",
              "--host-resolver-rules=MAP * ~NOTFOUND",
              "--no-first-run",
              "--no-default-browser-check",
              f"--user-data-dir={stage / 'profile'}",
              "--no-pdf-header-footer",
              f"--print-to-pdf={staged_pdf}",
              staged_html.as_uri(),
          ]
          for headless in ("--headless=new", "--headless"):
              staged_pdf.unlink(missing_ok=True)
              try:
                  result = subprocess.run(
                      [base[0], headless, *base[1:]],
                      stdout=subprocess.PIPE,
                      stderr=subprocess.PIPE,
                      timeout=180,
                  )
              except (OSError, subprocess.TimeoutExpired) as exc:
                  diagnostics = str(exc)
                  continue
              diagnostics = result.stderr.decode("utf-8", "replace")[-2000:]
              if result.returncode == 0 and staged_pdf.is_file():
                  data = staged_pdf.read_bytes()
                  if data.startswith(b"%PDF-") and data.rstrip().endswith(b"%%EOF") and count_pages(staged_pdf):
                      # Staging is on the destination filesystem: a failed render
                      # never removes or truncates an existing report.
                      staged_pdf.replace(out_path)
                      return 0
          sys.stderr.write(diagnostics)
      return 3
  
  
  def count_pages(pdf_path: Path) -> int:
      return len(re.findall(rb"/Type\s*/Page\b", pdf_path.read_bytes()))
  
  
  def render_png(pdf_path: Path, png_dir: Path) -> None:
      try:
          import pypdfium2 as pdfium  # type: ignore
      except ImportError:
          print(
              "pypdfium2 не установлен: страницы не отрисованы. Просмотрите PDF другим способом "
              "(встроенный просмотр PDF агента, pdftoppm, системный просмотрщик)."
          )
          return
      png_dir.mkdir(parents=True, exist_ok=True)
      pdf = pdfium.PdfDocument(str(pdf_path))
      for index in range(len(pdf)):
          pdf[index].render(scale=1.3).to_pil().save(png_dir / f"page{index + 1}.png")
      print(f"Страницы отрисованы: {png_dir} ({len(pdf)} шт.)")
  
  
  def main() -> int:
      args = parse_args()
      html_path = Path(args.html).expanduser().resolve()
      if not html_path.is_file():
          print(f"Ошибка: нет файла {html_path}", file=sys.stderr)
          return 2
  
      problems, missing = check_html(html_path)
      if missing:
          print("Ошибка: не найдены изображения: " + ", ".join(missing), file=sys.stderr)
      if problems:
          print("HTML не прошёл проверку:", file=sys.stderr)
          for problem in problems:
              print(f"- {problem}", file=sys.stderr)
      if missing or problems:
          return 1
      if args.check_only:
          print("Проверка пройдена.")
          return 0
  
      chrome = find_chrome(args.chrome)
      if not chrome:
          print(
              "Ошибка: не найден Chrome, Chromium или Edge. Укажите путь через --chrome или CHROME_PATH.",
              file=sys.stderr,
          )
          return 2
  
      out_path = Path(args.out).expanduser().resolve() if args.out else html_path.with_suffix(".pdf")
      out_path.parent.mkdir(parents=True, exist_ok=True)
      code = print_pdf(chrome, html_path, out_path)
      if code:
          print("Ошибка: браузер не смог напечатать PDF.", file=sys.stderr)
          return code
  
      print(f"PDF собран: {out_path} (страниц: {count_pages(out_path)}, {out_path.stat().st_size // 1024} КБ)")
      if args.png_dir:
          render_png(out_path, Path(args.png_dir).expanduser().resolve())
      print("Просмотрите каждую страницу перед отправкой пользователю (SKILL.md, шаг 12, п. 8).")
      return 0
  
  
  if __name__ == "__main__":
      sys.path.insert(0, str(Path(__file__).resolve().parent))
      from research_common import configure_utf8_stdio
  
      configure_utf8_stdio()
      raise SystemExit(main())
  

• scripts/ca/russian_trusted_root_ca.pem2.0 KBfe7ad30756dbupstream ↗

  -----BEGIN CERTIFICATE-----
  MIIFwjCCA6qgAwIBAgICEAAwDQYJKoZIhvcNAQELBQAwcDELMAkGA1UEBhMCUlUx
  PzA9BgNVBAoMNlRoZSBNaW5pc3RyeSBvZiBEaWdpdGFsIERldmVsb3BtZW50IGFu
  ZCBDb21tdW5pY2F0aW9uczEgMB4GA1UEAwwXUnVzc2lhbiBUcnVzdGVkIFJvb3Qg
  Q0EwHhcNMjIwMzAxMjEwNDE1WhcNMzIwMjI3MjEwNDE1WjBwMQswCQYDVQQGEwJS
  VTE/MD0GA1UECgw2VGhlIE1pbmlzdHJ5IG9mIERpZ2l0YWwgRGV2ZWxvcG1lbnQg
  YW5kIENvbW11bmljYXRpb25zMSAwHgYDVQQDDBdSdXNzaWFuIFRydXN0ZWQgUm9v
  dCBDQTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAMfFOZ8pUAL3+r2n
  qqE0Zp52selXsKGFYoG0GM5bwz1bSFtCt+AZQMhkWQheI3poZAToYJu69pHLKS6Q
  XBiwBC1cvzYmUYKMYZC7jE5YhEU2bSL0mX7NaMxMDmH2/NwuOVRj8OImVa5s1F4U
  zn4Kv3PFlDBjjSjXKVY9kmjUBsXQrIHeaqmUIsPIlNWUnimXS0I0abExqkbdrXbX
  YwCOXhOO2pDUx3ckmJlCMUGacUTnylyQW2VsJIyIGA8V0xzdaeUXg0VZ6ZmNUr5Y
  Ber/EAOLPb8NYpsAhJe2mXjMB/J9HNsoFMBFJ0lLOT/+dQvjbdRZoOT8eqJpWnVD
  U+QL/qEZnz57N88OWM3rabJkRNdU/Z7x5SFIM9FrqtN8xewsiBWBI0K6XFuOBOTD
  4V08o4TzJ8+Ccq5XlCUW2L48pZNCYuBDfBh7FxkB7qDgGDiaftEkZZfApRg2E+M9
  G8wkNKTPLDc4wH0FDTijhgxR3Y4PiS1HL2Zhw7bD3CbslmEGgfnnZojNkJtcLeBH
  BLa52/dSwNU4WWLubaYSiAmA9IUMX1/RpfpxOxd4Ykmhz97oFbUaDJFipIggx5sX
  ePAlkTdWnv+RWBxlJwMQ25oEHmRguNYf4Zr/Rxr9cS93Y+mdXIZaBEE0KS2iLRqa
  OiWBki9IMQU4phqPOBAaG7A+eP8PAgMBAAGjZjBkMB0GA1UdDgQWBBTh0YHlzlpf
  BKrS6badZrHF+qwshzAfBgNVHSMEGDAWgBTh0YHlzlpfBKrS6badZrHF+qwshzAS
  BgNVHRMBAf8ECDAGAQH/AgEEMA4GA1UdDwEB/wQEAwIBhjANBgkqhkiG9w0BAQsF
  AAOCAgEAALIY1wkilt/urfEVM5vKzr6utOeDWCUczmWX/RX4ljpRdgF+5fAIS4vH
  tmXkqpSCOVeWUrJV9QvZn6L227ZwuE15cWi8DCDal3Ue90WgAJJZMfTshN4OI8cq
  W9E4EG9wglbEtMnObHlms8F3CHmrw3k6KmUkWGoa+/ENmcVl68u/cMRl1JbW2bM+
  /3A+SAg2c6iPDlehczKx2oa95QW0SkPPWGuNA/CE8CpyANIhu9XFrj3RQ3EqeRcS
  AQQod1RNuHpfETLU/A2gMmvn/w/sx7TB3W5BPs6rprOA37tutPq9u6FTZOcG1Oqj
  C/B7yTqgI7rbyvox7DEXoX7rIiEqyNNUguTk/u3SZ4VXE2kmxdmSh3TQvybfbnXV
  4JbCZVaqiZraqc7oZMnRoWrXRG3ztbnbes/9qhRGI7PqXqeKJBztxRTEVj8ONs1d
  WN5szTwaPIvhkhO3CO5ErU2rVdUr89wKpNXbBODFKRtgxUT70YpmJ46VVaqdAhOZ
  D9EUUn4YaeLaS8AjSF/h7UkjOibNc4qVDiPP+rkehFWM66PVnP1Msh93tc+taIfC
  EYVMxjh8zNbFuoc7fzvvrFILLe7ifvEIUqSVIC/AzplM/Jxw7buXFeGP1qVCBEHq
  391d/9RAfaZ12zkwFsl+IKwE/OZxW8AHa9i1p4GO0YSNuczzEm4=
  -----END CERTIFICATE-----
  

• scripts/ca/russian_trusted_sub_ca.pem2.4 KB1fbc4a47ea47upstream ↗

  -----BEGIN CERTIFICATE-----
  MIIG6DCCBNCgAwIBAgICEAUwDQYJKoZIhvcNAQELBQAwcDELMAkGA1UEBhMCUlUx
  PzA9BgNVBAoMNlRoZSBNaW5pc3RyeSBvZiBEaWdpdGFsIERldmVsb3BtZW50IGFu
  ZCBDb21tdW5pY2F0aW9uczEgMB4GA1UEAwwXUnVzc2lhbiBUcnVzdGVkIFJvb3Qg
  Q0EwHhcNMjQwNzE1MTI1MDQxWhcNMjkwNzE5MTI1MDQxWjBvMQswCQYDVQQGEwJS
  VTE/MD0GA1UECgw2VGhlIE1pbmlzdHJ5IG9mIERpZ2l0YWwgRGV2ZWxvcG1lbnQg
  YW5kIENvbW11bmljYXRpb25zMR8wHQYDVQQDDBZSdXNzaWFuIFRydXN0ZWQgU3Vi
  IENBMIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEA1j0rkZECOt1S8o7I
  JY+4YKAxuEa5xaHKHXT2EpkuC/0krqMOjUy2oPIRNgR5g8X0Jl6jamxeGLc4Q1tf
  ju6or9oSRYThIUhRsFDQNBiBBEXoBgWxTfiKB2eyT97+pz5TBtBiRCPaLGRHYLRb
  9Jz2HkJlxbtNPjtDrF5DPHym+mZ1M1z3hIQYAqJwLpsEBnsw/VxWMlxqHoeewd0h
  uJMd71KQ5vOKlz7KrIZ6EobNNa6wItuvsfj3kYCK7O78uLHGXXFxdr8Hae9lMUmC
  8F7AFwa+bO1LRlTlqW7rE3rLf+jj70N01N8T3o22v14YBaFBWQWncAVYD2JuL3tH
  252+kdNOERf1fLbLRigJAbd+hOhWYlNf963TFDgnNPliHNIW72SygVBnI2V3JwO1
  dp1hVKpK/zt8ziGdHW4gmOLTsH50YKdR4jNqUgQv4wASlKn9OpN6zHYc5G8h86fY
  BM+zxE5ikGI+I/vIqBuI0eaDU92AWN/YjFLpu8tMu9kLRSCf1vug6FIfDPWVo7iP
  ac/SI2v8jnnpaW7ph/Pz3WkzaG7ZZJsfFs+8dploWc6LOoDtbFBhMdGMxu024msC
  0PSjZb5ODXPIaO2NsA7fMiAtZcoK6anTUJh4zOP/stA9qsJGNxdrEmiPXSmBZY/N
  Y0wkZgZ6JTDhw7038bPvctkblJkCAwEAAaOCAYswggGHMB0GA1UdDgQWBBR3Pdk5
  r0K93FvKduru/c4+YSkwXzAfBgNVHSMEGDAWgBTh0YHlzlpfBKrS6badZrHF+qws
  hzAOBgNVHQ8BAf8EBAMCAYYwEgYDVR0TAQH/BAgwBgEB/wIBADCBmAYIKwYBBQUH
  AQEEgYswgYgwQAYIKwYBBQUHMAKGNGh0dHA6Ly9udWMtY2RwLnZvc2tob2QucnUv
  Y2RwL3Jvb3RjYV9zc2xfcnNhMjAyMi5jcnQwRAYIKwYBBQUHMAKGOGh0dHA6Ly9u
  dWMtY2RwLmRpZ2l0YWwuZ292LnJ1L2NkcC9yb290Y2Ffc3NsX3JzYTIwMjIuY3J0
  MIGFBgNVHR8EfjB8MDqgOKA2hjRodHRwOi8vbnVjLWNkcC52b3NraG9kLnJ1L2Nk
  cC9yb290Y2Ffc3NsX3JzYTIwMjIuY3JsMD6gPKA6hjhodHRwOi8vbnVjLWNkcC5k
  aWdpdGFsLmdvdi5ydS9jZHAvcm9vdGNhX3NzbF9yc2EyMDIyLmNybDANBgkqhkiG
  9w0BAQsFAAOCAgEAmsINXtQ7wwUWvIeOr80MdJS/5G4xhyZOVEmeUorThquT672y
  cCg3XCxc4fwbiZqSSbBqntQ7RtiTAKMYMvBageKoVHbzz+R4jX01tKcTx8cDePrz
  dJ73bLNUorE7RU9QsW4KyiUeRmjMDV23AUlEvuQFTwgkHXvbac1BBdPn9CrssQuF
  5EGohZKcQPFiAAc4SHbRNhlr7uAwgpc/erzI9EAcvA6BVAXcVKoeGpV01uexUgZ6
  St5RP9UmDWNA7T4yVXWJ233N0Q8bl+6AswINQ3PosPu6yQQHQjr65YS06epK+AeI
  6j+oGR4xI7EhTQhQvaobnGmX/8QQ7XDRYCP2HXYxiffnn/CfZ/BVyKLYeY1ZipjE
  nzqdQIC2+Q3WtY8jsVRQMP38WFRmtsIt5snehnPTs5bKGVIcYzj3o3Ex/K7agEz0
  zAJ0JR5ivXZOvNkT0g9x1v+S1IkU3e/nX1a+tpRquMtnHX0L2lXArNHUbaOO9EJt
  d57WaIpofV5cVhhwShOgAuBc9UMJF3/n4t4RKiPxtsK8P67gcmphMhslj7AMYrYM
  ej2NvQZY4m3ub3CPC/PrTjDONvb+8g5xrKtxBjYqC74HSB4dg9G3WimSDUuP2Su6
  G2y2TUeyJuCvCLz289VoO0vg7cNdMobE3KCqAiiNhN2VBFxHAUKmUoRcRdw=
  -----END CERTIFICATE-----
  

• scripts/familysearch.py3.3 KBe0b9afc53411upstream ↗

  #!/usr/bin/env python3
  """Search names in saved FamilySearch evidence fragments."""
  
  from __future__ import annotations
  
  import argparse
  import re
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).resolve().parent))  # `python -I` omits the script folder
  
  from familysearch_browser import browser_import
  from research_common import configure_utf8_stdio, SourceError, outside_skill, output_json, read_json
  
  
  def compile_patterns(config=None, surnames=(), beacons=()):
      config = config or {}
      if not isinstance(config, dict) or set(config) - {"_comment", "surnames", "beacons"}:
          raise SourceError("Неизвестные поля конфигурации поиска.")
      definitions = []
      for layer, extra in (("surnames", surnames), ("beacons", beacons)):
          values = config.get(layer, {})
          if not isinstance(values, dict):
              raise SourceError(f"{layer}: ожидается объект метка → выражение.")
          definitions.extend((layer, label, pattern) for label, pattern in values.items())
          definitions.extend((layer, pattern, pattern) for pattern in extra)
      result, labels = [], set()
      for layer, label, pattern in definitions:
          if not isinstance(label, str) or not label.strip() or label in labels or not isinstance(pattern, str) or not pattern:
              raise SourceError("Пустая/повторная метка или некорректное выражение.")
          labels.add(label)
          try:
              result.append((layer, label, re.compile(pattern, re.IGNORECASE)))
          except re.error as exc:
              raise SourceError(f"Некорректное Python regex для метки {label}.") from exc
      return result
  
  
  def parser():
      root = argparse.ArgumentParser(allow_abbrev=False, description="FamilySearch: локальный поиск по именам и тексту сохранённых фрагментов индекса.")
      commands = root.add_subparsers(dest="command", required=True)
      browser = commands.add_parser("browser-import", allow_abbrev=False, help="Разобрать сохранённые точечные фрагменты индекса")
      browser.add_argument("files", nargs="+")
      browser.add_argument("--surnames", nargs="+", default=[])
      browser.add_argument("--beacons", nargs="+", default=[])
      browser.add_argument("--config")
      browser.add_argument("--out", help="JSON вне папки навыка; по умолчанию stdout")
      return root
  
  
  def run(args):
      if args.out:
          outside_skill(args.out)
      config = read_json(args.config) if args.config else {}
      patterns = compile_patterns(config, args.surnames, args.beacons)
      result = browser_import(args.files, patterns)
      output_json(result, args.out)
      return 0
  
  
  def main(argv=None):
      args = parser().parse_args(argv)
      try:
          return run(args)
      except (SourceError, OSError, ValueError, TypeError, KeyError) as exc:
          message = str(exc) if isinstance(exc, SourceError) else "Некорректный входной файл или ошибка локального чтения/записи."
          print("Ошибка: " + message, file=sys.stderr)
          return 1
  
  
  if __name__ == "__main__":
      configure_utf8_stdio()
      raise SystemExit(main())
  

• scripts/familysearch_browser.py6.8 KB72236ec3d7aaupstream ↗

  """Analyze targeted FamilySearch evidence fragments."""
  
  from __future__ import annotations
  
  import hashlib
  import json
  import re
  import urllib.parse
  from pathlib import Path
  
  from research_common import SourceError, timestamp
  
  
  def fs_url(value, kind=None):
      parsed = urllib.parse.urlsplit(value or "")
      if parsed.scheme != "https" or parsed.netloc not in ("www.familysearch.org", "familysearch.org"):
          raise SourceError("Нужна HTTPS-ссылка FamilySearch без учётных данных в URL.")
      if kind and not re.fullmatch(r"/ark:/61903/" + re.escape(kind) + r":[^/]+", parsed.path):
          raise SourceError("Проверьте устойчивую ссылку записи или изображения FamilySearch.")
      return parsed
  
  
  def snapshot_rows(snapshot, pages):
      rows, scope = snapshot.get("rows"), snapshot.get("scope")
      if not isinstance(rows, list) or not isinstance(scope, dict):
          raise SourceError("Браузерный фрагмент должен содержать url, scope и rows.")
      fs_url(snapshot.get("url"))
      state = snapshot.get("index_state", "indexed" if rows else "unresolved")
      if state not in ("indexed", "not-indexed", "unresolved", "blocked"):
          raise SourceError("Неизвестное состояние индекса.")
      if snapshot.get("loading"):
          state = "unresolved"
      evidence = snapshot.get("state_evidence", "")
      if not isinstance(evidence, str):
          raise SourceError("state_evidence должен содержать наблюдённый текст источника.")
      if state == "not-indexed" and not evidence.strip():
          raise SourceError("Отсутствие индекса требует явного сообщения источника в state_evidence.")
      if (state == "indexed") != bool(rows):
          raise SourceError("Состояние индекса не соответствует строкам; не сохраняйте незагруженные строки как проверенные.")
      context = {"url": snapshot["url"], "scope": scope, "captured_at": snapshot.get("captured_at", ""),
                 "index_state": state, "state_evidence": evidence}
      pages.append(context)
      return [(row, context) for row in rows]
  
  
  def browser_import(files, patterns):
      candidates, registry, hits, pages, seen = [], [], {}, [], {}
      rows, inputs = [], []
      for path in files:
          path = Path(path).expanduser()
          content = path.read_bytes()
          snapshot = json.loads(content.decode("utf-8-sig"))
          inputs.append({"path": str(path.resolve()), "sha256": hashlib.sha256(content).hexdigest()})
          if not isinstance(snapshot, dict):
              raise SourceError("Некорректный браузерный фрагмент.")
          rows.extend(snapshot_rows(snapshot, pages))
      for row, context in rows:
          if not isinstance(row, dict) or not isinstance(row.get("name"), str) or not row["name"].strip() or not isinstance(row.get("text"), str):
              raise SourceError("В браузерной строке нужны исходные name, text и ссылка.")
          observed_id = row.get("observed_id")
          if observed_id is not None and (not isinstance(observed_id, str) or not observed_id.strip()):
              raise SourceError("Локальный observed_id должен быть непустой строкой.")
          parsed = fs_url(row.get("url"), "3:1" if observed_id else "1:1")
          if row.get("image_url"):
              fs_url(row["image_url"], "3:1")
          ark = "" if observed_id else parsed.path.split("1:1:")[-1]
          rid = "browser-observation:" + observed_id if observed_id else ark
          original = {k: row[k] for k in ("name", "text", "url", "image_url", "people", "observed_id") if k in row}
          if rid in seen:
              if seen[rid] != original:
                  raise SourceError("Браузерные строки одной записи различаются; сохраните версии раздельно.")
              continue
          seen[rid] = original
          people = row.get("people") or [{"name": row["name"]}]
          if not isinstance(people, list) or any(not isinstance(p, dict) or not isinstance(p.get("name"), str) or not p["name"] for p in people):
              raise SourceError("Некорректный список наблюдённых имён.")
          persons = {(f"observed-{i}" if row.get("people") else "principal"): {"name": p["name"], "names": [p["name"]], "ark": ark if i == 1 else "",
                      "principal": i == 1, "indexed_label": p.get("label", "")}
                     for i, p in enumerate(people, 1)}
          matches = []
          for layer, label, regex in patterns:
              person_matches = [pid for pid, person in persons.items() if regex.search(person["name"])]
              if regex.search(row["text"]) or person_matches:
                  for pid in person_matches or [None]:
                      match = {"rid": rid, "pid": pid, "name": persons[pid]["name"] if pid else row["name"],
                               "principal": persons[pid]["principal"] if pid else None, "ark": ark,
                               "layer": layer, "label": label, "match_scope": "observed-fragment"}
                      hits.setdefault(label, []).append(match)
                      matches.append(match)
          registry.append({"rid": rid, "name": row["name"], "ark": ark, "indexed_text": row["text"], "url": row["url"]})
          if patterns and not matches:
              continue
          record = {"persons": persons,
                    "browser_text": row["text"], "browser_url": row["url"],
                    "identifier_kind": "local-browser-observation" if observed_id else "record-ark",
                    "captured_at": context["captured_at"], "raw": {"browser_row": original}}
          location = {"page_url": context["url"], "image_url": row.get("image_url", ""), "scope": context["scope"]}
          candidates.append({"rid": rid, "record": record, "locations": [location], "matches": matches})
      report = {"complete": False, "scope": "targeted-browser-fragments", "observed_rows": len(seen),
                "matched_rows": len(candidates), "pages": pages, "index_completeness": "not-established",
                "unresolved_pages": [p["url"] for p in pages if p["index_state"] == "unresolved"],
                "not_indexed_pages": [p["url"] for p in pages if p["index_state"] == "not-indexed"],
                "blocked_pages": [p["url"] for p in pages if p["index_state"] == "blocked"]}
      return {"source": "familysearch", "transport": "browser", "analyzed_at": timestamp(), "registry": registry,
              "hits": hits, "candidates": candidates, "inputs": inputs, "query": [{"layer": layer, "label": label, "pattern": regex.pattern}
              for layer, label, regex in patterns], "coverage": report}
  

• scripts/gwar.py17 KB93d4b01fb40fupstream ↗

  #!/usr/bin/env python3
  """Find WWI source records and locality clusters without inferring kinship."""
  
  from __future__ import annotations
  
  import argparse
  import difflib
  import re
  import ssl
  import sys
  import xml.etree.ElementTree as ET
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).resolve().parent))  # `python -I` omits the script folder
  
  from research_common import (configure_utf8_stdio, SourceClient, SourceError, fingerprint, outside_skill, output_json,
                               read_json, save_json, state_lock, timestamp)
  
  BASE = "https://gwar.mil.ru"
  KINDS = {"reports": "chelovek_donesenie", "hospitals": "chelovek_gospital", "awards": "chelovek_nagrazhdenie", "pows": "chelovek_plen"}
  
  
  def source_client():
      # Extra certificates apply only to this client, never to the system trust store.
      context = ssl.create_default_context()
      for name in ("russian_trusted_root_ca.pem", "russian_trusted_sub_ca.pem"):
          path = Path(__file__).resolve().parent / "ca" / name
          if not path.is_file():
              raise SourceError("Отсутствует сертификат GWAR из scripts/ca; проверка TLS не отключается.")
          context.load_verify_locations(cafile=str(path))
      return SourceClient(BASE, context=context, referer=BASE + "/heroes/")
  
  
  def query_body(fields, offset, size, entities, builder="Heroes"):
      return {"indices": ["gwar"], "entities": entities, "queryFields": fields,
              "filterFields": {}, "facetField": None, "from": offset, "size": size, "builderType": builder}
  
  
  def fetch_page(client, fields, offset, size, entities, builder="Heroes"):
      data = client.request("/gt_data/?builder=" + builder, query_body(fields, offset, size, entities, builder))
      if isinstance(data, dict) and "response" in data:
          data = data["response"]
      box = data.get("hits") if isinstance(data, dict) else None
      if not isinstance(box, dict) or not isinstance(box.get("hits"), list):
          raise SourceError("GWAR: неожиданный формат выдачи.")
      total = box.get("total")
      if isinstance(total, dict):
          if total.get("relation", "eq") != "eq":
              raise SourceError("GWAR не сообщил точное число результатов; полнота неизвестна.")
          total = total.get("value")
      if type(total) is not int or total < 0:
          raise SourceError("GWAR: отсутствует точный счётчик записей.")
      for hit in box["hits"]:
          if not isinstance(hit, dict) or not hit.get("_id") or not isinstance(hit.get("_source"), dict):
              raise SourceError("GWAR: карточка без ID или исходных полей.")
      return box["hits"], total
  
  
  def hit_key(hit):
      return "|".join(str(hit.get(field, "")) for field in ("_index", "_type", "_id"))
  
  
  def collect(client, fields, entities, limit, checkpoint=None):
      identity = {"source": "gwar", "query": fields, "entities": entities}
      state = {**identity, "fingerprint": fingerprint(identity), "total": None, "hits": [],
               "next_offset": 0, "started_at": timestamp(), "captured_at": timestamp(), "coverage": {"complete": False}}
      if checkpoint and Path(checkpoint).exists():
          state = read_json(checkpoint)
          if state.get("fingerprint") != fingerprint(identity):
              raise SourceError("Сохранённая выдача GWAR относится к другому запросу.")
          if not isinstance(state.get("hits"), list) or type(state.get("next_offset")) is not int:
              raise SourceError("Некорректное состояние выдачи GWAR.")
          if limit < len(state["hits"]):
              raise SourceError("Лимит меньше уже сохранённой выдачи; выберите другой файл или прежний лимит.")
      hits = {hit_key(hit): hit for hit in state["hits"]}
      if len(hits) != len(state["hits"]):
          raise SourceError("В контрольной точке GWAR есть повторные ID.")
      blocked, incomplete_reason = None, ""
      try:
          if state["total"] is None or state["next_offset"] < state["total"]:
              client.request("/heroes/", as_json=False)
          while len(hits) < limit and (state["total"] is None or state["next_offset"] < state["total"]):
              requested_size = min(1000, limit - len(hits))
              page, total = fetch_page(client, fields, state["next_offset"], requested_size, entities)
              if len(page) > requested_size:
                  raise SourceError("GWAR вернул больше запрошенного объёма; проверьте формат API.")
              if state["total"] is not None and state["total"] != total:
                  raise SourceError("Число результатов GWAR изменилось; проверьте выдачу новым запросом.")
              state["total"] = total
              state["captured_at"] = timestamp()
              previous_count = len(hits)
              for hit in page:
                  key = hit_key(hit)
                  if key in hits and hits[key] != hit:
                      raise SourceError("Содержимое карточки GWAR изменилось во время получения выдачи.")
                  hits[key] = hit
              state["next_offset"] += len(page)
              state["hits"] = list(hits.values())
              if not page or len(hits) == previous_count:
                  if len(hits) < total:
                      incomplete_reason = "empty-or-repeated-page-before-total"
                  break
              if checkpoint:
                  save_json(checkpoint, state)
      except SourceError as exc:
          blocked = {"status": exc.status, "message": str(exc)}
      finally:
          complete = state["total"] is not None and len(hits) == state["total"] and not blocked and not incomplete_reason
          state.update({"hits": list(hits.values()), "updated_at": timestamp(), "blocked": blocked,
                        "coverage": {"complete": complete, "retrieved_records": len(hits), "reported_total": state["total"],
                                     "limit": limit, "reason": incomplete_reason or (("limit-reached" if len(hits) >= limit else "source-window-incomplete") if not complete and not blocked else ""),
                                     "scope": "returned-source-records-not-population-census"}})
          if checkpoint:
              save_json(checkpoint, state)
      return state
  
  
  def parse_scan(xml):
      xml = str(xml or "")
      image = re.search(r"[A-Za-z0-9_\-/]+\.(?:jpe?g|png|tiff?)", xml, flags=re.IGNORECASE)
      path = image.group(0) if image else None
      zones = []
      try:
          root = ET.fromstring(xml)
          for element in root.iter():
              if element.tag.rsplit("}", 1)[-1] == "s":
                  zones.append(dict(element.attrib))
      except ET.ParseError:
          # Some source payloads are XML fragments rather than a full document.
          for match in re.finditer(r"<s\s+([^>]+?)/?>", xml):
              zones.append(dict(re.findall(r'([\w:]+)\s*=\s*"([^"]*)"', match.group(1))))
      return {"image_path": path, "image_url": BASE + "/cdn/imagesfww/" + path.lstrip("/") if path else None,
              "zones": zones, "raw": xml}
  
  
  def pages(client, page_ids):
      client.request("/heroes/", as_json=False)
      result = []
      for page_id in dict.fromkeys(page_ids):
          hits, total = fetch_page(client, {"linkedPages": [int(page_id)]}, 0, 100, ["stranitsa"], "HeroesStranitsa")
          result.append({"pages_id": page_id, "total": total, "complete": len(hits) == total,
                         "records": [{"id": hit["_id"], "source": hit["_source"],
                                      **parse_scan(hit["_source"].get("obraz_s_oblastyami", ""))} for hit in hits]})
      return {"source": "gwar-pages", "captured_at": timestamp(), "pages": result,
              "coverage": {"complete": all(item["complete"] for item in result), "scope": "selected-pages"}}
  
  
  PLACE_UNITS = (("gubernia", re.compile(r"\b(?:губ(?:ерния)?|обл(?:асть)?)\.?$")),
                 ("uezd", re.compile(r"\b(?:уезд|у)\.?$")),
                 ("volost", re.compile(r"\bвол(?:ость)?\.?$")))
  SETTLEMENT_TYPE = re.compile(r"^(?:селение|село|деревня|дер|хутор|местечко|колония|слобода|н\.\s?п|с|д|х|м|к|сл)\.?\s+")
  
  
  def place_parts(source):
      """Split the card's place into units and the settlement, keeping each string as written."""
      parts = {unit: str(source.get("birth_place_" + unit) or "").strip() for unit, _ in PLACE_UNITS}
      rest = []
      for token in (item.strip() for item in str(source.get("birth_place") or "").split(",")):
          if not token:
              continue
          unit = next((name for name, pattern in PLACE_UNITS if pattern.search(token.lower())), None)
          if unit and not parts[unit]:
              parts[unit] = token
          elif not unit:
              rest.append(token)
      parts["settlement"] = ", ".join(rest)
      return parts
  
  
  def place_key(name):
      name = SETTLEMENT_TYPE.sub("", name.lower().replace("ё", "е").strip())
      return re.sub(r"[^a-zа-яіїєґăâîșțş]", "", name)
  
  
  def similarity(left, right):
      left, right = place_key(left), place_key(right)
      if not left or not right:
          return 0.0
      skeleton = [re.sub(r"[аеиоуыэюяіїєйăâîьъ]", "", value) for value in (left, right)]
      scores = [difflib.SequenceMatcher(None, left, right).ratio()]
      if all(skeleton):
          scores.append(difflib.SequenceMatcher(None, *skeleton).ratio())
      return round(max(scores), 2)
  
  
  def places(states, target=None):
      """Group settlement spellings of saved cards by gubernia, uezd and volost; no network access."""
      cards, inputs = {}, []
      for name, state in states:
          if (not isinstance(state, dict) or state.get("source", "gwar") != "gwar"
                  or not isinstance(state.get("hits"), list)):
              raise SourceError(f"{name}: это не выдача gwar.py search/volost.")
          coverage = state.get("coverage")
          if not isinstance(coverage, dict):
              raise SourceError(f"{name}: отсутствуют сведения о полноте выдачи GWAR.")
          inputs.append({"file": name, "query": state.get("query"),
                         "complete": coverage.get("complete") is True})
          for hit in state["hits"]:
              if (not isinstance(hit, dict) or not hit.get("_id")
                      or not isinstance(hit.get("_source"), dict)):
                  raise SourceError(f"{name}: карточка GWAR без ID или исходных полей.")
              key = hit_key(hit)
              if key in cards and cards[key] != hit:
                  raise SourceError(f"{name}: версии одной карточки GWAR различаются; сравните исходные файлы.")
              cards[key] = hit
      if not inputs:
          raise SourceError("Укажите хотя бы одну сохранённую выдачу GWAR.")
      groups = {}
      for hit in cards.values():
          parts = place_parts(hit.get("_source") or {})
          group = groups.setdefault((parts["gubernia"], parts["uezd"], parts["volost"]), {})
          entry = group.setdefault(parts["settlement"], {"as_written": parts["settlement"], "cards": 0, "card_ids": [], "card_keys": []})
          entry["cards"] += 1
          entry["card_ids"].append(hit["_id"])
          entry["card_keys"].append(hit_key(hit))
      result = []
      for (gubernia, uezd, volost), settlements in groups.items():
          items = list(settlements.values())
          for item in items:
              if target:
                  item["similarity"] = similarity(target, item["as_written"])
          items.sort(key=lambda item: (-item.get("similarity", 0), -item["cards"], item["as_written"]))
          result.append({"gubernia": gubernia, "uezd": uezd, "volost": volost, "settlements": items})
      result.sort(key=lambda group: (-max((item.get("similarity", 0) for item in group["settlements"]), default=0),
                                     group["gubernia"], group["uezd"], group["volost"]))
      return {"source": "gwar-places", "captured_at": timestamp(), "target": target, "inputs": inputs,
              "cards": len(cards), "groups": result,
              "coverage": {"complete": all(item["complete"] for item in inputs),
                           "scope": "spellings-in-saved-cards-not-settlement-list"}}
  
  
  def positive(value):
      number = int(value)
      if number < 1:
          raise argparse.ArgumentTypeError("Нужно положительное число.")
      return number
  
  
  def parser():
      root = argparse.ArgumentParser(description="Документы Первой мировой: поиск человека и сбор по местности.")
      commands = root.add_subparsers(dest="command", required=True)
      search = commands.add_parser("search", help="Поиск по имени, месту или части")
      for flag in ("surname", "given", "patronymic", "event-place", "regiment", "birth-place", "gubernia"):
          help_text = ("Населённый пункт в поле birth_place сайта (место жительства / призыва; не доказанное место рождения)"
                       if flag == "birth-place" else None)
          search.add_argument("--" + flag, help=help_text)
      volost = commands.add_parser("volost", help="Собрать найденные записи по волости и губернии")
      volost.add_argument("--volost", required=True, help="Волость в поле birth_place_volost сайта")
      volost.add_argument("--gubernia", required=True)
      for command in (search, volost):
          command.add_argument("--kind", nargs="+", choices=("all", *KINDS), default=["all"])
          command.add_argument("--max-records", type=positive, default=1000)
          command.add_argument("--out", required=True, help="JSON вне папки навыка; также контрольная точка")
      spellings = commands.add_parser("places", help="Сравнить написания местностей в сохранённых выдачах (без сети)")
      spellings.add_argument("--input", nargs="+", required=True, help="JSON-выдачи search/volost")
      spellings.add_argument("--target", help="Неопознанное название, с которым сравнить написания")
      spellings.add_argument("--out", help="JSON вне папки навыка; без него вывод в консоль")
      page = commands.add_parser("page", help="Найти скан и координаты строк по pages_id")
      page.add_argument("--pages", nargs="+", type=positive, required=True)
      page.add_argument("--out", required=True)
      return root
  
  
  def main(argv=None):
      args = parser().parse_args(argv)
      try:
          if args.command == "places":
              if args.out and Path(args.out).expanduser().resolve() in {Path(name).expanduser().resolve() for name in args.input}:
                  raise SourceError("Выходной файл не должен заменять исходную выдачу GWAR.")
              result = places([(name, read_json(name)) for name in args.input], args.target)
              output_json(result, outside_skill(args.out) if args.out else None)
              return 0 if result["coverage"]["complete"] else 1
          path = outside_skill(args.out)
          if args.command == "page":
              result = pages(source_client(), args.pages)
              output_json(result, path)
          else:
              if args.command == "volost":
                  fields = {"birth_place_volost": args.volost, "birth_place_gubernia": args.gubernia}
              else:
                  mapping = {"surname": "last_name", "given": "first_name", "patronymic": "middle_name",
                             "event_place": "event_place", "regiment": "military_unit_name",
                             "birth_place": "birth_place", "gubernia": "birth_place_gubernia"}
                  fields = {field: getattr(args, key) for key, field in mapping.items() if getattr(args, key)}
                  if not fields:
                      raise SourceError("Укажите имя, местность или воинскую часть.")
              kinds = list(KINDS) if "all" in args.kind else list(dict.fromkeys(args.kind))
              with state_lock(str(path) + ".lock"):
                  result = collect(source_client(), fields, [KINDS[kind] for kind in kinds], args.max_records, path)
          return 0 if result["coverage"]["complete"] else 1
      except (SourceError, OSError, ValueError, TypeError, KeyError) as exc:
          print("Ошибка: " + (str(exc) if isinstance(exc, SourceError) else "Некорректный файл или ошибка локального ввода/вывода."), file=sys.stderr)
          return 1
  
  
  if __name__ == "__main__":
      configure_utf8_stdio()
      raise SystemExit(main())
  

• scripts/hash_artifact.py1.8 KB7b1522172018upstream ↗

  #!/usr/bin/env python3
  """Compute a SHA-256 checksum and basic metadata for a local artifact."""
  
  from __future__ import annotations
  
  import argparse
  import hashlib
  import json
  import sys
  from datetime import datetime, timezone
  from pathlib import Path
  
  
  def sha256_file(path: Path) -> str:
      digest = hashlib.sha256()
      with path.open("rb") as handle:
          for chunk in iter(lambda: handle.read(1024 * 1024), b""):
              digest.update(chunk)
      return digest.hexdigest()
  
  
  def parse_args() -> argparse.Namespace:
      parser = argparse.ArgumentParser(description="Вычисляет SHA-256 локального артефакта.")
      parser.add_argument("path", help="Путь к файлу")
      parser.add_argument("--json", action="store_true", help="Вывести JSON")
      return parser.parse_args()
  
  
  def main() -> int:
      args = parse_args()
      path = Path(args.path).expanduser().resolve()
      if not path.exists():
          print(f"Ошибка: файл не найден: {path}", file=sys.stderr)
          return 2
      if not path.is_file():
          print(f"Ошибка: путь не является файлом: {path}", file=sys.stderr)
          return 2
  
      stat = path.stat()
      result = {
          "path": str(path),
          "filename": path.name,
          "size_bytes": stat.st_size,
          "modified_at_utc": datetime.fromtimestamp(stat.st_mtime, timezone.utc)
          .replace(microsecond=0)
          .isoformat(),
          "sha256": sha256_file(path),
      }
      if args.json:
          print(json.dumps(result, ensure_ascii=False, indent=2))
      else:
          for key, value in result.items():
              print(f"{key}: {value}")
      return 0
  
  
  if __name__ == "__main__":
      sys.path.insert(0, str(Path(__file__).resolve().parent))
      from research_common import configure_utf8_stdio
  
      configure_utf8_stdio()
      raise SystemExit(main())
  

• scripts/init_case.py4.7 KBd267db49b791upstream ↗

  #!/usr/bin/env python3
  """Create a portable case directory from the bundled assets."""
  
  from __future__ import annotations
  
  import argparse
  import shutil
  import sys
  from datetime import datetime, timezone
  from pathlib import Path
  
  ASSET_FILES = {
      "case-intake.template.yaml": "case-intake.yaml",
      "person-register.csv": "person-register.csv",
      "artifact-register.csv": "artifact-register.csv",
      "claims-register.csv": "claims-register.csv",
      "relationship-register.csv": "relationship-register.csv",
      "search-log.csv": "search-log.csv",
      "final-report.template.md": "final-report.md",
      "research-state.template.md": "research-state.md",
  }
  PHOTOS_README = "photos-readme.template.md"
  
  
  def quote_yaml(value: str) -> str:
      return '"' + value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", " ") + '"'
  
  
  def parse_args() -> argparse.Namespace:
      parser = argparse.ArgumentParser(
          description="Создаёт структуру дела для исследования родственных связей."
      )
      parser.add_argument("--case-dir", required=True, help="Директория нового дела")
      parser.add_argument("--research-question", required=True, help="Главный исследовательский вопрос")
      parser.add_argument("--subject-label", default="", help="Краткая метка основного человека")
      parser.add_argument(
          "--force",
          action="store_true",
          help="Перезаписать только файлы шаблонов; не удаляет иные файлы",
      )
      return parser.parse_args()
  
  
  def main() -> int:
      args = parse_args()
      script_dir = Path(__file__).resolve().parent
      assets_dir = script_dir.parent / "assets"
      case_dir = Path(args.case_dir).expanduser().resolve()
  
      if case_dir == script_dir.parent or script_dir.parent in case_dir.parents:
          print(
              "Ошибка: храните дело вне папки навыка, например ~/kinship-cases/my-case.",
              file=sys.stderr,
          )
          return 2
  
      missing = [name for name in ASSET_FILES if not (assets_dir / name).is_file()]
      if missing:
          print(
              "Ошибка: в skill отсутствуют шаблоны: " + ", ".join(missing),
              file=sys.stderr,
          )
          return 2
  
      if case_dir.exists() and not case_dir.is_dir():
          print(f"Ошибка: {case_dir} существует и не является директорией.", file=sys.stderr)
          return 2
  
      case_dir.mkdir(parents=True, exist_ok=True)
      conflicts = [dest for dest in ASSET_FILES.values() if (case_dir / dest).exists()]
      if conflicts and not args.force:
          print(
              "Ошибка: целевые файлы уже существуют: "
              + ", ".join(conflicts)
              + ". Продолжайте работу с существующим делом. Для намеренного сброса шаблонов есть --force.",
              file=sys.stderr,
          )
          return 2
  
      for source_name, dest_name in ASSET_FILES.items():
          shutil.copyfile(assets_dir / source_name, case_dir / dest_name)
  
      photos_dir = case_dir / "photos"
      photos_dir.mkdir(exist_ok=True)
      photos_readme = photos_dir / "README.md"
      if not photos_readme.exists() and (assets_dir / PHOTOS_README).is_file():
          shutil.copyfile(assets_dir / PHOTOS_README, photos_readme)
  
      intake_path = case_dir / "case-intake.yaml"
      text = intake_path.read_text(encoding="utf-8")
      created_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
      case_id = case_dir.name
      replacements = {
          'case_id: ""': f"case_id: {quote_yaml(case_id)}",
          'created_at: ""': f"created_at: {quote_yaml(created_at)}",
          'research_question: ""': f"research_question: {quote_yaml(args.research_question)}",
          'subject_label: ""': f"subject_label: {quote_yaml(args.subject_label)}",
          '    label: ""': f"    label: {quote_yaml(args.subject_label)}",
      }
      for old, new in replacements.items():
          if old not in text:
              print(f"Ошибка: маркер шаблона не найден: {old}", file=sys.stderr)
              return 3
          text = text.replace(old, new, 1)
      intake_path.write_text(text, encoding="utf-8")
  
      print(f"Дело создано: {case_dir}")
      for dest_name in ASSET_FILES.values():
          print(f"- {dest_name}")
      print("- photos/README.md")
      return 0
  
  
  if __name__ == "__main__":
      sys.path.insert(0, str(Path(__file__).resolve().parent))
      from research_common import configure_utf8_stdio
  
      configure_utf8_stdio()
      raise SystemExit(main())
  

• scripts/register_search.py18 KBb489c47809cfupstream ↗

  #!/usr/bin/env python3
  """Register source-search leads (FamilySearch, GWAR, or any site via a manual JSON) in an existing case without asserting identities."""
  
  from __future__ import annotations
  
  import argparse
  import csv
  import hashlib
  import io
  import json
  import re
  import sys
  import urllib.parse
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).resolve().parent))  # `python -I` omits the script folder
  
  from research_common import (configure_utf8_stdio, SourceError, atomic_bytes, fingerprint, outside_skill,
                               output_json, state_lock, timestamp)
  
  
  def read_register(path, id_field):
      with path.open(encoding="utf-8-sig", newline="") as handle:
          reader = csv.DictReader(handle)
          fields = reader.fieldnames
          if not fields or id_field not in fields:
              raise SourceError(f"Некорректный реестр {path.name}.")
          rows = list(reader)
      if any(None in row or any(value is None for value in row.values()) for row in rows):
          raise SourceError(f"Повреждена строка CSV: {path.name}.")
      ids = [row[id_field] for row in rows]
      if any(not value for value in ids) or len(set(ids)) != len(ids):
          raise SourceError(f"Пустые или повторные ID в {path.name}.")
      return fields, rows
  
  
  def save_register(path, fields, rows):
      stream = io.StringIO(newline="")
      writer = csv.DictWriter(stream, fieldnames=fields)
      writer.writeheader()
      writer.writerows(rows)
      atomic_bytes(path, stream.getvalue().encode("utf-8"))
  
  
  def next_id(rows, field, prefix):
      numbers = [int(match.group(1)) for row in rows if (match := re.fullmatch(re.escape(prefix) + r"(\d+)", row[field]))]
      return f"{prefix}{max(numbers, default=0) + 1:04d}"
  
  
  def encode(value):
      return json.dumps(value, ensure_ascii=False, sort_keys=True)
  
  
  def fs_items(data):
      candidates = data.get("candidates")
      if not isinstance(candidates, list):
          raise SourceError("Передайте результат локального анализа фрагментов FamilySearch.")
      for candidate in candidates:
          record, rid = candidate.get("record"), candidate.get("rid")
          if not rid or not isinstance(record, dict) or not isinstance(record.get("persons"), dict) or not record["persons"]:
              raise SourceError("Некорректный кандидат FamilySearch.")
          people = record["persons"]
          principal = next((pid for pid, person in people.items() if person.get("principal")), next(iter(people)))
          ark = people[principal].get("ark", "")
          if record.get("browser_url"):
              url = record["browser_url"]
          elif ark.startswith("https://www.familysearch.org/"):
              url = ark
          elif ark and not urllib.parse.urlsplit(ark).scheme:
              url = "https://www.familysearch.org/ark:/61903/1:1:" + urllib.parse.quote(ark, safe="")
          else:
              url = ""  # No public locator observed; retain the local source snapshot.
          locations = candidate.get("locations", [])
          reference = {"RID": rid if record.get("identifier_kind") != "local-browser-observation" else "",
                       "local_observation_id": rid if record.get("identifier_kind") == "local-browser-observation" else "",
                       "locations": locations}
          names = list(dict.fromkeys(name for person in people.values() for name in (person.get("names") or [person.get("name", "")]) if name))
          yield {"id": str(rid), "snapshot": record, "locations": locations,
                 "captured_at": record.get("captured_at", ""), "source_title": "FamilySearch — индекс записи " + str(rid),
                 "source_url_or_path": url, "repository_or_archive": "FamilySearch; архив оригинала уточнить",
                 "archival_reference": encode(reference),
                 "names_as_written": "; ".join(names),
                 "summary": "Кандидат из текстового фрагмента индекса; личность и родство требуют проверки."}
  
  
  # Report/hospital card routes reported by the contributor in PR #50 (2026-10-27);
  # POW card route reported by the contributor in PR #64 (2026-10-28).
  GWAR_CARD_PAGES = ("chelovek_donesenie", "chelovek_gospital", "chelovek_plen")
  
  
  def gwar_items(data):
      if not isinstance(data.get("hits"), list):
          raise SourceError("Для GWAR передайте результат search или volost.")
      for hit in data["hits"]:
          if not isinstance(hit, dict) or not hit.get("_id") or not isinstance(hit.get("_source"), dict):
              raise SourceError("Некорректная карточка GWAR.")
          source = hit["_source"]
          identity = "|".join(str(hit.get(field, "")) for field in ("_index", "_type", "_id"))
          name = " ".join(str(source.get(field) or "") for field in ("last_name", "first_name", "middle_name")).strip()
          url = "https://gwar.mil.ru/heroes/"
          if hit.get("_type") in GWAR_CARD_PAGES and re.fullmatch(r"[0-9]+", str(hit["_id"])):
              url += f"{hit['_type']}{hit['_id']}/"
          yield {"id": identity, "snapshot": hit, "captured_at": data.get("captured_at", ""),
                 "source_title": "GWAR — " + (name or str(hit["_id"])),
                 "source_url_or_path": url, "repository_or_archive": "GWAR; архив оригинала уточнить",
                 "archival_reference": encode({"card": identity, "documents_pages": source.get("documents_pages", {}),
                                                "military_unit_name": source.get("military_unit_name", "")}),
                 "record_date": str(source.get("event_date_from") or ""),
                 # Report cards usually leave birth_place empty and fill volost/uezd/gubernia.
                 "place_as_written": ", ".join(str(source[field]) for field in (
                     "birth_place", "birth_place_volost", "birth_place_uezd", "birth_place_gubernia") if source.get(field)),
                 "names_as_written": name,
                 "summary": "Карточка Первой мировой; место и одноимённые лица требуют проверки по документу."}
  
  
  MANUAL_TEXT_FIELDS = ("source_url_or_path", "repository_or_archive", "archival_reference", "record_date",
                        "place_as_written", "names_as_written", "language_script", "summary")
  MANUAL_ITEM_KEYS = {"id", "source_title", "snapshot", *MANUAL_TEXT_FIELDS}
  
  
  def manual_items(data):
      """Validate a hand-described search in any source and yield items shaped like fs_items/gwar_items."""
      domain, query = data.get("source_or_domain"), data.get("query")
      if not isinstance(domain, str) or not domain.strip():
          raise SourceError("Для manual укажите непустой source_or_domain.")
      if not (isinstance(query, dict) and query) and not (isinstance(query, str) and query.strip()):
          raise SourceError("Для manual укажите query: непустой объект или строку.")
      if "captured_at" in data and not isinstance(data["captured_at"], str):
          raise SourceError("captured_at должен быть строкой.")
      if not isinstance(data.get("items"), list):
          raise SourceError("Для manual items должен быть списком (пустым, если находок нет).")
      for number, item in enumerate(data["items"], 1):
          if not isinstance(item, dict):
              raise SourceError(f"manual: элемент {number} должен быть объектом.")
          unknown = sorted(set(item) - MANUAL_ITEM_KEYS)
          if unknown:
              raise SourceError(f"manual: элемент {number} содержит неизвестные поля: {', '.join(unknown)}.")
          for field in ("id", "source_title"):
              if not isinstance(item.get(field), str) or not item[field].strip():
                  raise SourceError(f"manual: у элемента {number} нужно непустое поле {field}.")
          for field in MANUAL_TEXT_FIELDS:
              if field in item and not isinstance(item[field], str):
                  raise SourceError(f"manual: поле {field} элемента {number} должно быть строкой.")
          if "snapshot" in item and not isinstance(item["snapshot"], dict):
              raise SourceError(f"manual: поле snapshot элемента {number} должно быть объектом.")
          yield {"id": item["id"], "snapshot": {"source_or_domain": domain.strip(), **item}, "captured_at": data.get("captured_at", ""),
                 "source_title": item["source_title"],
                 **{field: item.get(field, "") for field in MANUAL_TEXT_FIELDS if field not in ("repository_or_archive", "summary")},
                 "repository_or_archive": item.get("repository_or_archive") or "уточнить по источнику",
                 "summary": item.get("summary") or "Наводка из ручного поиска; личность и родство требуют проверки."}
  
  
  def without_capture(value):
      if isinstance(value, dict):
          return {key: without_capture(item) for key, item in value.items() if key != "captured_at"}
      if isinstance(value, list):
          return [without_capture(item) for item in value]
      return value
  
  
  def snapshot_key(source, identity, snapshot):
      snapshot = without_capture(snapshot)
      if source == "gwar" and isinstance(snapshot, dict):
          # Search ranking varies between name and locality queries of the same card.
          snapshot = {key: value for key, value in snapshot.items() if key not in ("_score", "sort", "_explanation")}
      return fingerprint({"source": source, "id": identity, "snapshot": snapshot})
  
  
  def register(case, input_path, goal):
      case = outside_skill(case)
      input_bytes = Path(input_path).expanduser().read_bytes()
      data = json.loads(input_bytes.decode("utf-8-sig"))
      if not isinstance(data, dict) or data.get("source") not in ("familysearch", "gwar", "manual"):
          raise SourceError("Нужен JSON локального анализа FamilySearch, поиска GWAR или ручного поиска (source: manual).")
      source = data["source"]
      # Validate the complete input before changing any register.
      items = list({"familysearch": fs_items, "gwar": gwar_items, "manual": manual_items}[source](data))
      coverage = data.get("coverage")
      if source == "manual" and coverage is None:
          # Unknown coverage stays partial; do not force the researcher to invent completeness.
          coverage = {"complete": False}
      if not isinstance(coverage, dict) or type(coverage.get("complete")) is not bool:
          raise SourceError("В результате отсутствует отчёт о покрытии.")
      if source == "familysearch":
          if any(coverage.get(key) for key in ("unresolved_pages", "blocked_pages")):
              coverage = {**coverage, "complete": False}
      for name in ("person-register.csv", "claims-register.csv", "relationship-register.csv", "case-intake.yaml"):
          if not (case / name).is_file():
              raise SourceError("Сначала создайте дело через init_case.py.")
      with state_lock(case / ".register-search.lock"):
          artifact_path, log_path = case / "artifact-register.csv", case / "search-log.csv"
          artifact_fields, artifacts = read_register(artifact_path, "artifact_id")
          log_fields, logs = read_register(log_path, "search_id")
          required_artifact = {"artifact_id", "access_notes", "originality", "directness", "status", "sha256", "independence_group"}
          required_log = {"search_id", "filters", "artifact_ids", "result_status"}
          if not required_artifact.issubset(artifact_fields) or not required_log.issubset(log_fields):
              raise SourceError("Формат реестров не соответствует шаблонам навыка.")
          existing_keys = {}
          artifact_ids, added = [], 0
          for item in items:
              key = snapshot_key(source, item["id"], item["snapshot"])
              marker = "source_key=" + key
              previous = existing_keys.get(key) or next((row for row in artifacts if marker in row.get("access_notes", "")), None)
              if previous:
                  artifact_ids.append(previous["artifact_id"])
                  continue
              # Save the actual source payload, not just a surname-match excerpt.
              payload = {"source": source, "source_record_id": item["id"], "record": item["snapshot"],
                         "locations": item.get("locations", []), "registered_at": timestamp()}
              snapshot = case / "sources" / source / (key + ".json")
              # Reject symlinks that would put private case material inside the skill.
              outside_skill(snapshot)
              content = (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
              if snapshot.exists():
                  content = snapshot.read_bytes()
                  existing = json.loads(content)
                  if (existing.get("source_record_id") != item["id"]
                          or without_capture(existing.get("record")) != without_capture(item["snapshot"])):
                      raise SourceError("Сохранённый артефакт не соответствует исходной записи.")
              else:
                  atomic_bytes(snapshot, content)
              artifact_id = next_id(artifacts, "artifact_id", "A-")
              row = {field: "" for field in artifact_fields}
              row.update({key: value for key, value in item.items() if key in row})
              row.update({"artifact_id": artifact_id, "captured_at": item["captured_at"] or timestamp(),
                          "source_url_or_path": item["source_url_or_path"] or str(snapshot),
                          "source_type": "unknown" if source == "manual" else "index",
                          "originality": "unknown" if source == "manual" else "index",
                          "directness": "lead-only", "status": "lead-only",
                          "provenance_level": ("Ручная заметка; происхождение первоисточника не проверено"
                                               if source == "manual" else "источник индекса установлен; оригинал не проверен"),
                          "access_notes": f"{marker}; snapshot={snapshot.relative_to(case)}",
                          "sha256": hashlib.sha256(content).hexdigest(),
                          "reliability_notes": "Совпадение — кандидат. Независимость не установлена; назначьте группу исходного документа после проверки."
                          + (" Дата исходного получения неизвестна; captured_at — дата регистрации." if not item["captured_at"] else "")})
              artifacts.append(row)
              existing_keys[key] = row
              artifact_ids.append(artifact_id)
              added += 1
          artifact_ids = list(dict.fromkeys(artifact_ids))
          complete = coverage["complete"] and not data.get("blocked")
          status = "found" if artifact_ids else "not-found"
          if not complete:
              status = "blocked" if data.get("blocked") else "partial"
          log_key = fingerprint({"source": source, "query": data.get("query"), "coverage": coverage,
                                 "artifacts": artifact_ids, "goal": goal, "input": fingerprint(data)})
          log_marker = "search_key=" + log_key
          new_log = not any(log_marker in row.get("filters", "") for row in logs)
          if new_log:
              row = {field: "" for field in log_fields}
              row.update({"search_id": next_id(logs, "search_id", "S-"), "timestamp": timestamp(), "research_goal": goal,
                          "query": encode(data.get("query", {"apid": data.get("apid")})),
                          "source_or_domain": {"familysearch": "familysearch.org", "gwar": "gwar.mil.ru"}.get(source) or data["source_or_domain"],
                          "filters": encode({"key": log_marker, "coverage": coverage, "entities": data.get("entities"),
                                             "input": str(Path(input_path).expanduser().resolve()), "sha256": hashlib.sha256(input_bytes).hexdigest()}),
                          "result_status": status, "artifact_ids": ";".join(artifact_ids),
                          "blocker": encode(data["blocked"]) if data.get("blocked") else "",
                          "next_step": "Проверить оригиналы, различить кандидатов и установить независимость; покрытие ограничено указанным набором."})
              logs.append(row)
          # A rerun after an interruption between these writes reuses source_key and search_key.
          save_register(artifact_path, artifact_fields, artifacts)
          save_register(log_path, log_fields, logs)
          return {"added_artifacts": added, "artifact_ids": artifact_ids, "added_search_log": new_log,
                  "result_status": status, "claims_created": 0, "persons_created": 0, "relationships_created": 0}
  
  
  def main(argv=None):
      parser = argparse.ArgumentParser(description="Записать поисковые находки в существующие реестры как непроверенные артефакты.")
      parser.add_argument("--case-dir", required=True)
      parser.add_argument("--input", required=True, help="Результат FamilySearch browser-import, GWAR search/volost или JSON ручного поиска (source: manual)")
      parser.add_argument("--research-goal", required=True)
      args = parser.parse_args(argv)
      try:
          output_json(register(args.case_dir, args.input, args.research_goal))
          return 0
      except (SourceError, OSError, ValueError, TypeError, KeyError) as exc:
          print("Ошибка: " + (str(exc) if isinstance(exc, SourceError) else "Некорректный файл или ошибка локального ввода/вывода."), file=sys.stderr)
          return 1
  
  
  if __name__ == "__main__":
      configure_utf8_stdio()
      raise SystemExit(main())
  

• scripts/research_common.py7.2 KBa406807bdeefupstream ↗

  """Local checkpoints and bounded, read-only HTTP for source search helpers."""
  
  from __future__ import annotations
  
  import hashlib
  import http.cookiejar
  import json
  import os
  import re
  import socket
  import ssl
  import sys
  import tempfile
  import time
  import urllib.error
  import urllib.parse
  import urllib.request
  from contextlib import contextmanager
  from datetime import datetime, timezone
  from pathlib import Path
  
  
  def configure_utf8_stdio():
      """Keep CLI output readable when an agent captures a legacy-encoded pipe."""
      for stream in (sys.stdout, sys.stderr):
          reconfigure = getattr(stream, "reconfigure", None)
          if reconfigure is not None:
              reconfigure(encoding="utf-8")
  
  
  class SourceError(Exception):
      """A source or checkpoint could not be read reliably."""
  
      def __init__(self, message, status=None):
          super().__init__(message)
          self.status = status
  
  
  def timestamp():
      return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
  
  
  def fingerprint(value):
      data = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
      return hashlib.sha256(data.encode("utf-8")).hexdigest()
  
  
  def read_json(path):
      with Path(path).expanduser().open(encoding="utf-8-sig") as handle:
          return json.load(handle)
  
  
  def atomic_bytes(path, content):
      path = Path(path).expanduser().resolve()
      path.parent.mkdir(parents=True, exist_ok=True)
      fd, temporary = tempfile.mkstemp(prefix="." + path.name + ".", dir=path.parent)
      try:
          with os.fdopen(fd, "wb") as handle:
              handle.write(content)
              handle.flush()
              os.fsync(handle.fileno())
          os.replace(temporary, path)
      finally:
          Path(temporary).unlink(missing_ok=True)
  
  
  def save_json(path, value):
      atomic_bytes(path, (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8"))
  
  
  def output_json(value, path=None):
      if path:
          save_json(path, value)
      else:
          print(json.dumps(value, ensure_ascii=False, indent=2))
  
  
  def outside_skill(path):
      path = Path(path).expanduser().resolve()
      root = Path(__file__).resolve().parent.parent
      if path == root or root in path.parents:
          raise SourceError("Храните результаты исследования вне папки навыка.")
      return path
  
  
  @contextmanager
  def state_lock(path):
      """Exclude concurrent writers. Stale locks need an explicit local check."""
      path = Path(path)
      path.parent.mkdir(parents=True, exist_ok=True)
      try:
          fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
      except FileExistsError as exc:
          raise SourceError(f"Состояние занято: {path}. Проверьте процесс перед удалением lock-файла.") from exc
      try:
          with os.fdopen(fd, "w", encoding="utf-8") as handle:
              handle.write(str(os.getpid()))
          yield
      finally:
          path.unlink(missing_ok=True)
  
  
  class SameOriginRedirect(urllib.request.HTTPRedirectHandler):
      def redirect_request(self, request, response, code, message, headers, new_url):
          old = urllib.parse.urlsplit(request.full_url)
          new = urllib.parse.urlsplit(new_url)
          if (old.scheme, old.netloc) != (new.scheme, new.netloc):
              raise SourceError("Источник перенаправляет на другой адрес; проверьте вход в браузере.", code)
          if code in (301, 302, 303) and request.get_method() == "POST":
              raise SourceError("Источник перенаправляет запрос; проверьте формат API и доступ.", code)
          return super().redirect_request(request, response, code, message, headers, new_url)
  
  
  class SourceClient:
      """Requests stay on one source. No retries for access denial or rate limits."""
  
      def __init__(self, base, context=None, interval=0.4, referer=None):
          self.base = base.rstrip("/")
          self.referer = referer or self.base + "/"
          self.interval = interval
          self.last_request = 0.0
          self.cookies = http.cookiejar.CookieJar()
          self.opener = urllib.request.build_opener(
              SameOriginRedirect(), urllib.request.HTTPCookieProcessor(self.cookies),
              urllib.request.HTTPSHandler(context=context or ssl.create_default_context()),
          )
  
      def request(self, path, body=None, as_json=True):
          url = self.base + path
          # The only POSTs in these helpers are source searches, never tree mutations.
          data = None if body is None else json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
          headers = {
              "User-Agent": "ex-ussr-kinship-research/1.6 (source research)",
              "Accept": "application/json" if as_json else "text/html",
          }
          if body is not None:
              headers.update({"Content-Type": "application/json", "Referer": self.referer, "Origin": self.base})
          for attempt in range(3):
              time.sleep(max(0, self.interval - (time.monotonic() - self.last_request)))
              self.last_request = time.monotonic()
              request = urllib.request.Request(url, data=data, headers=headers)
              try:
                  with self.opener.open(request, timeout=45) as response:
                      content = response.read().decode("utf-8")
                  if re.search(r'<(?:iframe|div|form)[^>]*(?:smartcaptcha|g-recaptcha|captcha-container)', content, re.IGNORECASE):
                      raise SourceError("Источник показывает CAPTCHA; проверку проходит пользователь.", 403)
                  if not as_json:
                      return content
                  if content.lstrip().startswith("<"):
                      raise SourceError("Вместо данных получена веб-страница; проверьте вход или CAPTCHA в браузере.", 403)
                  try:
                      return json.loads(content)
                  except json.JSONDecodeError as exc:
                      raise SourceError("Источник вернул не JSON: возможен вход, CAPTCHA или изменение API.") from exc
              except urllib.error.HTTPError as exc:
                  if exc.code in (401, 403, 429):
                      raise SourceError(f"HTTP {exc.code}: запросы остановлены; проверьте доступ или лимит.", exc.code) from exc
                  if exc.code not in (500, 502, 503, 504) or attempt == 2:
                      raise SourceError(f"HTTP {exc.code}: источник недоступен.", exc.code) from exc
                  retry_after = exc.headers.get("Retry-After")
                  if retry_after:
                      raise SourceError("Источник просит повторить запрос позже; состояние сохранено.", exc.code) from exc
              except (urllib.error.URLError, TimeoutError, socket.timeout, OSError) as exc:
                  if attempt == 2:
                      raise SourceError("Сетевая ошибка или ошибка TLS; проверьте доступ к источнику.") from exc
              time.sleep(2 ** attempt)
          raise SourceError("Источник недоступен.")
  

• scripts/scan_montage.py8.1 KB80d5e40bcb7fupstream ↗

  #!/usr/bin/env python3
  """Собирает из одинаково устроенных сканов ленты-обзоры одной графы.
  
  Из каждого кадра вырезается одна и та же область (например, графа родителей
  в метрической книге или строка «Фамилия, имя и отчество домохозяина»
  в подворной карточке), при необходимости поворачивается, усиливается по
  контрасту и подписывается номером кадра. Полосы складываются в ленты по
  N кадров, чтобы просматривать сотни кадров быстро, а кандидатов затем
  открывать в полном разрешении. Требует Pillow (`pip install pillow`).
  """
  
  from __future__ import annotations
  
  import argparse
  import re
  import sys
  from pathlib import Path
  
  IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".tif", ".tiff"}
  
  
  def frame_number(path: Path) -> int | None:
      match = re.search(r"(\d+)(?!.*\d)", path.stem)
      return int(match.group(1)) if match else None
  
  
  def select_frames(directory: Path, first: int | None, last: int | None) -> list[Path]:
      frames = []
      for path in sorted(directory.iterdir()):
          if not path.is_file() or path.suffix.lower() not in IMAGE_SUFFIXES:
              continue
          number = frame_number(path)
          if number is None:
              continue
          if first is not None and number < first:
              continue
          if last is not None and number > last:
              continue
          frames.append(path)
      return sorted(frames, key=lambda p: (frame_number(p), p.name))
  
  
  def parse_box(value: str) -> tuple[float, float, float, float]:
      parts = [float(x) for x in value.split(",")]
      if len(parts) != 4:
          raise argparse.ArgumentTypeError("нужно четыре доли через запятую: x0,y0,x1,y1")
      x0, y0, x1, y1 = parts
      if not (0 <= x0 < x1 <= 1 and 0 <= y0 < y1 <= 1):
          raise argparse.ArgumentTypeError("доли должны быть в [0, 1], x0 < x1, y0 < y1")
      return x0, y0, x1, y1
  
  
  def make_strip(path: Path, box, half: str, rotate: int, width: int, contrast: float):
      from PIL import Image, ImageEnhance, ImageOps
  
      with Image.open(path) as source:
          image = ImageOps.exif_transpose(source).convert("L")
      w, h = image.size
      if half == "right" or (half == "auto" and w > h):
          image = image.crop((w // 2, 0, w, h))
      elif half == "left":
          image = image.crop((0, 0, w // 2, h))
      if rotate:
          image = image.rotate(rotate, expand=True)
      w, h = image.size
      x0, y0, x1, y1 = box
      pixels = (int(w * x0), int(h * y0), int(w * x1), int(h * y1))
      if pixels[2] <= pixels[0] or pixels[3] <= pixels[1]:
          raise ValueError(f"область после округления пуста: {path.name}")
      strip = image.crop(pixels)
      if contrast != 1.0:
          strip = ImageEnhance.Contrast(strip).enhance(contrast)
      height = max(1, round(width * strip.size[1] / strip.size[0]))
      return strip.resize((width, height))
  
  
  def build_sheets(frames, box, half, rotate, width, contrast, per_sheet, out_dir: Path, prefix: str):
      from PIL import Image, ImageDraw
  
      if "/" in prefix or "\\" in prefix:
          raise ValueError("--prefix должен быть префиксом имени, а не путём")
      numbers = [frame_number(path) for path in frames]
      if len(numbers) != len(set(numbers)):
          raise ValueError("номера кадров повторяются; выберите серию с однозначными номерами")
      chunks = [frames[start : start + per_sheet] for start in range(0, len(frames), per_sheet)]
      targets = [out_dir / f"{prefix}{frame_number(chunk[0]):04d}-{frame_number(chunk[-1]):04d}.jpg"
                 for chunk in chunks]
      for target in targets:
          if target.exists() or target.is_symlink():
              raise FileExistsError(f"файл уже существует, перезапись запрещена: {target}")
  
      label_width = 70
      written = []
      for chunk, target in zip(chunks, targets):
          rows = []
          for path in chunk:
              strip = make_strip(path, box, half, rotate, width, contrast)
              row = Image.new("L", (label_width + width, strip.size[1] + 4), 255)
              row.paste(strip, (label_width, 0))
              ImageDraw.Draw(row).text((4, 4), str(frame_number(path)), fill=0)
              rows.append(row)
          sheet = Image.new("L", (label_width + width, sum(r.size[1] for r in rows)), 255)
          y = 0
          for row in rows:
              sheet.paste(row, (0, y))
              y += row.size[1]
          with target.open("xb") as output:
              sheet.save(output, format="JPEG", quality=82)
          written.append(target)
      return written
  
  
  def parse_args(argv=None) -> argparse.Namespace:
      parser = argparse.ArgumentParser(description="Собирает ленты-обзоры одной графы из серии сканов.")
      parser.add_argument("directory", help="Папка с кадрами; номер кадра — последнее число в имени файла")
      parser.add_argument("--out", required=True, help="Папка для лент (создаётся при необходимости)")
      parser.add_argument("--box", type=parse_box, required=True, help="Область графы долями кадра после поворота: x0,y0,x1,y1")
      parser.add_argument("--first", type=int, help="Первый номер кадра")
      parser.add_argument("--last", type=int, help="Последний номер кадра")
      parser.add_argument("--half", choices=["none", "left", "right", "auto"], default="none",
                          help="Половина разворота; auto берёт правую половину у альбомных кадров")
      parser.add_argument("--rotate", type=int, choices=[0, 90, -90, 180], default=0, help="Поворот перед вырезкой, градусы")
      parser.add_argument("--per-sheet", type=int, default=12, help="Кадров на ленту (по умолчанию 12)")
      parser.add_argument("--width", type=int, default=1000, help="Ширина полосы в пикселях")
      parser.add_argument("--contrast", type=float, default=1.0, help="Усиление контраста (1.0 — без изменений)")
      parser.add_argument("--prefix", default="strip-", help="Префикс имён лент")
      return parser.parse_args(argv)
  
  
  def main(argv=None) -> int:
      args = parse_args(argv)
      try:
          import PIL  # noqa: F401
      except ImportError:
          print("Ошибка: нужен Pillow (pip install pillow)", file=sys.stderr)
          return 2
      directory = Path(args.directory).expanduser()
      if not directory.is_dir():
          print(f"Ошибка: папка не найдена: {directory}", file=sys.stderr)
          return 2
      if args.per_sheet < 1 or args.width < 50:
          print("Ошибка: --per-sheet должно быть ≥ 1, --width ≥ 50", file=sys.stderr)
          return 2
      frames = select_frames(directory, args.first, args.last)
      if not frames:
          print("Ошибка: в заданном диапазоне нет кадров", file=sys.stderr)
          return 1
      out_dir = Path(args.out).expanduser()
      out_dir.mkdir(parents=True, exist_ok=True)
      try:
          written = build_sheets(frames, args.box, args.half, args.rotate, args.width, args.contrast,
                                 args.per_sheet, out_dir, args.prefix)
      except (OSError, ValueError) as exc:
          print(f"Ошибка: {exc}", file=sys.stderr)
          return 2
      for path in written:
          print(path)
      print(f"Кадров: {len(frames)}; лент: {len(written)}", file=sys.stderr)
      return 0
  
  
  if __name__ == "__main__":
      sys.path.insert(0, str(Path(__file__).resolve().parent))
      from research_common import configure_utf8_stdio
  
      configure_utf8_stdio()
      raise SystemExit(main())
  

• references/bessarabia-moldova.md20 KB11609e9020c4upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/browser-and-captcha.md18 KB911879666111upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/changelog.md38 KBed5ee80a9daaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/chechnya-deportation.md28 KB733eea0ebed7upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/correspondence.md27 KB9084c4976feaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/evidence-protocol.md33 KB2f76fe2271e0upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/ex-ussr-locale-guide.md26 KB33a39de02101upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/familysearch-indexes.md9.0 KB3cf809283336upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/familysearch-mcp.md8.0 KB14567e68998fupstream ↗

  # FamilySearch MCP: подключённые инструменты исследования
  
  Читайте, если в среде подключён FamilySearch MCP или пользователь просит его настроить. Подходящее подключение использовать сразу в пределах задачи и прав доступа, без отдельного включения режима. Без него продолжать браузерный поиск и работу с имеющимися материалами.
  
  ## 1. Подключение и цель
  
  Обнаружить реальные инструменты и прочитать схемы нужных операций. Если доступна диагностика авторизации, сверить аккаунт, среду и права; отсутствие инструмента с конкретным именем вроде `auth_status` само по себе не блокирует доступные операции.
  
  Использовать выбранное пользователем древо и место результата. Подключение не меняет их и не расширяет поручение: наличие FamilySearch не переносит туда автоматически результат, предназначенный для локального GEDCOM или другого выбранного пользователем места. Неясность аккаунта или целевого древа блокирует соответствующую запись, а не всё исследование. Различать рабочую среду и тестовую: успешная операция в Beta не подтверждает изменение production.
  
  ## 2. Поиск и чтение
  
  Структурированный поиск, чтение профиля и просмотр архивного изображения имеют разные возможности. Проверять поддерживаемые фильтры и пагинацию; не передавать DGS, даты или родителей так, будто любой сервер их принимает. Если нужного фильтра или просмотра нет, выбрать браузерный маршрут по `familysearch-indexes.md` и `field-lessons.md`.
  
  Пустой или обрезанный ответ не устанавливает отсутствие записи: покрытие и загрузку оценивать по `search-result-validation.md`. Недоступный просмотр не означает, что нет документа, и не останавливает другие маршруты.
  
  Стабильные FamilySearch ID/ARK, запрос, время чтения, исходные поля, ссылки и фактически просмотренный объём сохранять в существующем деле. Древо и подсказки — исходные материалы; утверждения, разрешение личности, каждое родственное ребро и зависимость копий оценивать по `evidence-protocol.md`. Описание источника или Memory не заменяет чтение самого документа.
  
  ## 3. Разрешённые изменения
  
  Обычные изменения по поставленной задаче и существующему разрешению выполнять без повторного запроса на каждый шаг. Перед изменением прочитать целевую запись и проверить соответствие людей, фактов и связей. Использовать фактическую схему операции, сохранять основания и источники; не заменять чужие сведения молча и не добавлять второго родителя только из-за брака с первым.
  
  Если интерфейс не позволяет ясно выразить гипотезу или нужную область записи (например, отдельное приватное древо), сохранить результат в деле и оставить только этот перенос в очереди. Наличие общего инструмента записи не подтверждает поддержку выбранного древа; не использовать другой scope ради технического успеха. Не выдумывать жизненный статус, права или недостающие сведения ради выполнения операции.
  
  После записи перечитать результат, сохранить удалённый ID и соответствие локальному ID. После тайм-аута, обрыва или неоднозначного ответа сначала проверить, применено ли изменение; не повторять автоматически неидемпотентную запись. Код успеха не доказывает правильность содержания или вывода.
  
  Изменившуюся исследовательскую версию сверить с ранее внесёнными сведениями; не удалять людей и связи автоматически из-за их отсутствия в новом экспорте. Действовать по существующим правилам приватности и координации, сохраняя противоречия.
  
  ## 4. Доступ, приватность и настройка
  
  Авторизацию выполняет пользователь штатным способом подключения. Не извлекать cookies или токены и не обходить защиту; порядок входа и реальных ограничений — `browser-and-captcha.md`. Подключённый сервер не отменяет условия источника, лимиты и доступ к конкретному изображению.
  
  Данные живых, частные файлы и публикацию обрабатывать по `privacy-and-safety.md`, сохраняя уже данное разрешение. Не считать Memories или записи публичного древа приватными по умолчанию: фактическую видимость выбранной операции учитывать отдельно от цитирования. Не включать передачу файлов автоматически вместе с разрешением менять факты или связи.
  
  Настройку выполнять по запросу пользователя и документации выбранного клиента и сервера, сохраняя существующие настройки и ограничивая доступ задачей. Наличие MCP не требует установки конкретной сторонней реализации или смены браузера. Не предлагать извлечение браузерной сессии и маскировку клиента как способ подключения или обхода отказа.
  
  В отчёте отделять проверенные изменения от предложений и блокировок. `audit_evidence.py` проверяет локальные реестры, а не удалённое древо; итоговый список внесённого подтверждать чтением целевых записей.
  

• references/field-lessons.md81 KBf66fdc31e247upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/gedcom.md8.0 KBfb6fce252416upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/germany-forced-labour.md12 KBc032f0261850upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/government-portals.md11 KBb2d3b023a6aaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/habsburg-galicia-bukovina.md16 KBa781f28235e1upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/photo-search.md13 KBe89cdb20fc65upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/privacy-and-safety.md8.5 KBdd2ac6d3eeaaupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/research-coordination.md11 KB33220618ad9eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/search-result-validation.md13 KB987e11c39834upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/seed-sources.md224 KBcfb1737aa780upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/source-strategy.md71 KBd8882640cea9upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• references/ww1-search.md18 KBe7fe427726f8upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/artifact-register.csv323 B3319df50d0dfupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/cano-vital-record-request.template.docx5.3 KB5c96af4c30afupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/case-intake.template.yaml2.1 KB22edbd1b006bupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/claims-register.csv161 Bb823bff5206eupstream ↗

  claim_id,claim_text,subject_person_ids,claim_type,artifact_ids,supporting_independence_groups,contradicting_artifact_ids,status,confidence,rationale,next_action
  

• assets/final-report.template.md4.8 KBa9d2632ce1e3upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/person-register.csv291 B28b8f84a5a1dupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/photos-readme.template.md1.4 KBe03cd7541e77upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/relationship-register.csv194 B49ca36e8a67aupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/report.template.html34 KBeb000520c0b8upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/research-state.template.md1.3 KB5ed6014b8e25upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• assets/search-log.csv161 Bf6bbd6c21011upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• .gitignore148 Bea00585c60d2upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• docs/cover.jpg204 KB136f12ae0300upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/claim-dependence.yaml4.4 KB6f53f9e39c77upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/discovery-and-logic.yaml80 KB9b6cacb596aeupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/gedcom-storage.yaml7.8 KB4ef7970a26d7upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/marginal-notes.yaml2.5 KBdca38163fe02upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/next-research-questions.yaml4.6 KBe31a603ca5b7upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/register-counts.yaml3.9 KBa0ae9745c9c8upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/research-coordination.yaml6.3 KBd4572054740eupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/search-validation.yaml15 KB25485c809c3fupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/source-acquisition.yaml9.7 KBdce409d0ced3upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• evals/tree-and-parish.yaml7.6 KB4fb77806c631upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• LICENSE1.0 KB7ecc2cbcd580upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.az.md16 KBb36ec781246cupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.be.md22 KB217a35b3d6d5upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.de.md15 KBb1d54d79d15fupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.et.md14 KBf0931436ca7aupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.hy.md24 KB7ee2f418d7b6upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.ja.md19 KB49e916672133upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.ka.md33 KBcd51fa917dddupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.kk.md23 KB3032b6085ea9upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.ky.md23 KB24d49a62d7feupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.lt.md14 KBf16235a8e6b4upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.lv.md14 KB45ce15014d69upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.pl.md15 KBbcd7695144feupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.ro.md15 KBa3c72771b153upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.ru.md23 KBd65d277d718aupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.tg.md22 KB9f63ad9ca52aupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.tk.md16 KB743eb011cb7fupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.uk.md22 KB04b96a307175upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• README.uz.md15 KBefb50c061502upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• tests/test_audit_evidence.py18 KB533a7ebda233upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• tests/test_build_pdf.py2.3 KBc05783fe0ff3upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• tests/test_cano_template.py3.1 KB3c363ef965f4upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• tests/test_familysearch_local.py12 KB21921fc33d34upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• tests/test_scan_montage.py6.0 KB0b407712a770upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• tests/test_source_search.py31 KB47d74facd108upstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

Plain-text copy, no login: /skill-bundles/denissergeevitch-ex-ussr-kinship-research.md