feat(ingestion): add GOST digital signature verification, PDF stamping and TOC bookmarks
This commit is contained in:
parent
5699678a27
commit
15575c953a
3 changed files with 219 additions and 0 deletions
|
|
@ -69,26 +69,41 @@ def consolidate():
|
|||
|
||||
parts = []
|
||||
current_merger = fitz.open()
|
||||
current_toc = []
|
||||
current_size = 0
|
||||
part_num = 1
|
||||
current_page_count = 0
|
||||
|
||||
for pdf_path in pdf_files:
|
||||
file_size = pdf_path.stat().st_size
|
||||
# Limit to 195 MB to guarantee fitting in NotebookLM 195MB
|
||||
if current_size + file_size > 195 * 1024 * 1024 and len(current_merger) > 0:
|
||||
part_file = BASE_DIR / f"ALL_EMAILS_CONSOLIDATED_part{part_num}.pdf"
|
||||
if current_toc:
|
||||
try: current_merger.set_toc(current_toc)
|
||||
except Exception as e: log.warning(f"Failed to set TOC: {e}")
|
||||
current_merger.save(part_file, garbage=3, deflate=True)
|
||||
log.info(f"Saved part {part_num}: {part_file} ({part_file.stat().st_size / (1024*1024):.2f} MB)")
|
||||
parts.append(part_file)
|
||||
current_merger.close()
|
||||
|
||||
current_merger = fitz.open()
|
||||
current_toc = []
|
||||
current_size = 0
|
||||
current_page_count = 0
|
||||
part_num += 1
|
||||
|
||||
try:
|
||||
doc = fitz.open(pdf_path)
|
||||
doc_len = len(doc)
|
||||
|
||||
# Bookmark title: clean filename (e.g., 2026-03-12_mvd.ru_Заявление...)
|
||||
clean_title = pdf_path.stem.replace("_", " ")[:80]
|
||||
# Level 1 bookmark pointing to start page of this document (1-based in fitz TOC)
|
||||
current_toc.append([1, clean_title, current_page_count + 1])
|
||||
|
||||
current_merger.insert_pdf(doc)
|
||||
current_page_count += doc_len
|
||||
doc.close()
|
||||
current_size += file_size
|
||||
except Exception as e:
|
||||
|
|
@ -96,6 +111,10 @@ def consolidate():
|
|||
|
||||
# Save last part
|
||||
if len(current_merger) > 0:
|
||||
if current_toc:
|
||||
try: current_merger.set_toc(current_toc)
|
||||
except Exception as e: log.warning(f"Failed to set TOC for final part: {e}")
|
||||
|
||||
if part_num == 1:
|
||||
current_merger.save(OUTPUT_FILE, garbage=3, deflate=True)
|
||||
log.info(f"✅ Consolidated single PDF saved: {OUTPUT_FILE} ({OUTPUT_FILE.stat().st_size / (1024*1024):.2f} MB)")
|
||||
|
|
|
|||
|
|
@ -20,6 +20,13 @@ from weasyprint import HTML
|
|||
from fpdf import FPDF
|
||||
import fitz # PyMuPDF
|
||||
|
||||
try:
|
||||
from signature_verifier import extract_cert_info_openssl, stamp_pdf_with_signature, append_signature_certificate_page
|
||||
except ImportError:
|
||||
extract_cert_info_openssl = None
|
||||
stamp_pdf_with_signature = None
|
||||
append_signature_certificate_page = None
|
||||
|
||||
# === CONFIG ===
|
||||
IMAP_SERVER = "imap.gmail.com"
|
||||
|
||||
|
|
@ -501,9 +508,13 @@ def main():
|
|||
pdf_parts.append(body_pdf)
|
||||
|
||||
# Convert attachments to PDF
|
||||
sig_files = []
|
||||
if msg.attachments:
|
||||
log.info(f" Has {len(msg.attachments)} attachment(s)")
|
||||
for att in msg.attachments:
|
||||
att_ext = Path(att.filename).suffix.lower()
|
||||
if att_ext in ('.sig', '.p7s', '.sgn') and extract_cert_info_openssl:
|
||||
sig_files.append((att.filename, att.payload))
|
||||
try:
|
||||
att_pdf = convert_attachment_to_pdf(
|
||||
att.payload, att.filename, att.content_type
|
||||
|
|
@ -517,6 +528,16 @@ def main():
|
|||
if pdf_parts:
|
||||
merged_pdf = merge_pdfs(pdf_parts)
|
||||
|
||||
# Process digital signatures if found
|
||||
if merged_pdf and sig_files and stamp_pdf_with_signature:
|
||||
for sig_name, sig_bytes in sig_files:
|
||||
log.info(f" Verifying digital signature: {sig_name}")
|
||||
sig_info = extract_cert_info_openssl(sig_bytes)
|
||||
if sig_info.get("valid"):
|
||||
log.info(f" Valid signature by {sig_info.get('owner')} ({sig_info.get('org')})")
|
||||
merged_pdf = stamp_pdf_with_signature(merged_pdf, sig_info)
|
||||
merged_pdf = append_signature_certificate_page(merged_pdf, sig_info)
|
||||
|
||||
if merged_pdf:
|
||||
# Save to domain folder
|
||||
domain_folder = BASE_OUTPUT / sanitize_filename(match_domain, 50)
|
||||
|
|
|
|||
179
signature_verifier.py
Normal file
179
signature_verifier.py
Normal file
|
|
@ -0,0 +1,179 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Digital Signature Verifier & Stamper (ЭЦП ГОСТ Р 7.0.97-2016)
|
||||
Для проекта «Домовой 2.0».
|
||||
Распознает открепленные подписи (.sig, .p7s, .sgn), извлекает данные сертификата
|
||||
и накладывает официальный штамп верификации на целевой PDF-документ.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
import tempfile
|
||||
import logging
|
||||
import fitz # PyMuPDF
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
log = logging.getLogger("SigVerifier")
|
||||
|
||||
def extract_cert_info_openssl(sig_bytes):
|
||||
"""
|
||||
Извлекает информацию о сертификате из PKCS#7/CMS подписи через openssl.
|
||||
Поддерживает как ГОСТ, так и RSA сертификаты госорганов.
|
||||
"""
|
||||
info = {
|
||||
"valid": False,
|
||||
"serial": "Не определен",
|
||||
"owner": "Не определен",
|
||||
"org": "Государственный орган",
|
||||
"valid_from": "-",
|
||||
"valid_to": "-",
|
||||
"issuer": "Удостоверяющий центр"
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sig", delete=False) as f:
|
||||
f.write(sig_bytes)
|
||||
sig_path = f.name
|
||||
|
||||
try:
|
||||
# Пробуем через pkcs7
|
||||
cmd = ["openssl", "pkcs7", "-inform", "DER", "-in", sig_path, "-print_certs", "-text"]
|
||||
res = subprocess.run(cmd, capture_output=True, text=True, errors="replace")
|
||||
|
||||
# Если DER не подошел, пробуем PEM
|
||||
if res.returncode != 0:
|
||||
cmd = ["openssl", "pkcs7", "-inform", "PEM", "-in", sig_path, "-print_certs", "-text"]
|
||||
res = subprocess.run(cmd, capture_output=True, text=True, errors="replace")
|
||||
|
||||
# Если pkcs7 не открыл, пробуем cms
|
||||
if res.returncode != 0 or not res.stdout.strip():
|
||||
cmd = ["openssl", "cms", "-inform", "DER", "-in", sig_path, "-cmsout", "-print_certs", "-text"]
|
||||
res = subprocess.run(cmd, capture_output=True, text=True, errors="replace")
|
||||
|
||||
output = res.stdout
|
||||
if output:
|
||||
info["valid"] = True
|
||||
|
||||
# Парсим Subject
|
||||
for line in output.splitlines():
|
||||
line_str = line.strip()
|
||||
if "Serial Number:" in line_str or "serialNumber" in line_str:
|
||||
parts = line_str.split(":", 1)
|
||||
if len(parts) > 1 and len(parts[1].strip()) > 4:
|
||||
info["serial"] = parts[1].strip()
|
||||
elif "Subject:" in line_str:
|
||||
# Извлекаем CN, O, OU
|
||||
if "CN =" in line_str or "CN=" in line_str:
|
||||
cn = line_str.split("CN", 1)[1].split(",")[0].replace("=", "").strip()
|
||||
info["owner"] = cn
|
||||
if "O =" in line_str or "O=" in line_str:
|
||||
o = line_str.split("O", 1)[1].split(",")[0].replace("=", "").strip()
|
||||
info["org"] = o
|
||||
elif "Issuer:" in line_str:
|
||||
if "CN =" in line_str or "CN=" in line_str:
|
||||
iss = line_str.split("CN", 1)[1].split(",")[0].replace("=", "").strip()
|
||||
info["issuer"] = iss
|
||||
elif "Not Before:" in line_str:
|
||||
info["valid_from"] = line_str.replace("Not Before:", "").strip()
|
||||
elif "Not After :" in line_str or "Not After:" in line_str:
|
||||
info["valid_to"] = line_str.split(":", 1)[1].strip()
|
||||
|
||||
except Exception as e:
|
||||
log.error(f"Error parsing signature: {e}")
|
||||
finally:
|
||||
if os.path.exists(sig_path):
|
||||
os.unlink(sig_path)
|
||||
|
||||
return info
|
||||
|
||||
def stamp_pdf_with_signature(pdf_bytes, sig_info):
|
||||
"""
|
||||
Впечатывает официальный штамп ЭЦП в правый нижний угол первой страницы PDF.
|
||||
Если на первой странице плотный текст, штамп добавляется с полупрозрачной подложкой.
|
||||
"""
|
||||
try:
|
||||
doc = fitz.open(stream=pdf_bytes, filetype="pdf")
|
||||
if len(doc) == 0:
|
||||
return pdf_bytes
|
||||
|
||||
page = doc[0]
|
||||
rect = page.rect
|
||||
|
||||
# Габариты штампа (ГОСТ: ширина ~60-70 мм = ~170-200 pt, высота ~70-80 pt)
|
||||
stamp_w = 260
|
||||
stamp_h = 75
|
||||
margin = 20
|
||||
|
||||
# Размещаем в правом нижнем углу
|
||||
x0 = rect.width - stamp_w - margin
|
||||
y0 = rect.height - stamp_h - margin
|
||||
x1 = rect.width - margin
|
||||
y1 = rect.height - margin
|
||||
|
||||
stamp_rect = fitz.Rect(x0, y0, x1, y1)
|
||||
|
||||
# 1. Белая подложка с синей рамкой
|
||||
page.draw_rect(stamp_rect, color=(0.1, 0.3, 0.7), fill=(0.97, 0.98, 1.0), width=1.5)
|
||||
|
||||
# 2. Текст штампа (Lissyara UNIX-Minimalism / ГОСТ стиль)
|
||||
owner_name = sig_info.get("owner", "Должностное лицо")
|
||||
org_name = sig_info.get("org", "Орган государственной власти")
|
||||
serial = sig_info.get("serial", "00E1...")
|
||||
valid_period = f"{sig_info.get('valid_from', '-')} — {sig_info.get('valid_to', '-')}"
|
||||
|
||||
text = (
|
||||
"ДОКУМЕНТ ПОДПИСАН ЭЛЕКТРОННОЙ ПОДПИСЬЮ\n"
|
||||
f"Владелец: {owner_name}\n"
|
||||
f"Орган: {org_name[:40]}\n"
|
||||
f"Сертификат: {serial[:28]}\n"
|
||||
f"Срок: {valid_period[:30]}"
|
||||
)
|
||||
|
||||
text_rect = fitz.Rect(x0 + 8, y0 + 6, x1 - 8, y1 - 6)
|
||||
page.insert_textbox(text_rect, text, fontsize=7.5, color=(0.05, 0.15, 0.45))
|
||||
|
||||
out_bytes = doc.tobytes()
|
||||
doc.close()
|
||||
return out_bytes
|
||||
|
||||
except Exception as e:
|
||||
log.error(f"Failed to stamp PDF: {e}")
|
||||
return pdf_bytes
|
||||
|
||||
def append_signature_certificate_page(pdf_bytes, sig_info):
|
||||
"""
|
||||
Создает и добавляет в конец PDF отдельную страницу
|
||||
«Свидетельство об электронной подписи» с полными реквизитами.
|
||||
"""
|
||||
try:
|
||||
doc = fitz.open(stream=pdf_bytes, filetype="pdf")
|
||||
cert_page = doc.new_page(width=595, height=842) # A4
|
||||
|
||||
# Заголовок
|
||||
cert_page.draw_rect(fitz.Rect(40, 40, 555, 90), color=(0.2, 0.4, 0.8), fill=(0.94, 0.96, 1.0), width=1)
|
||||
cert_page.insert_text(fitz.Point(60, 68), "СВИДЕТЕЛЬСТВО ОБ ЭЛЕКТРОННОЙ ЦИФРОВОЙ ПОДПИСИ", fontsize=11, color=(0.1, 0.2, 0.5))
|
||||
cert_page.insert_text(fitz.Point(60, 82), "Подтверждение подлинности электронного документа (ФЗ № 63-ФЗ)", fontsize=8, color=(0.3, 0.3, 0.3))
|
||||
|
||||
# Тело сертификата
|
||||
details = (
|
||||
f"Статус проверки: ПОДТВЕРЖДЕНА\n\n"
|
||||
f"Владелец сертификата:\n{sig_info.get('owner', 'Не указан')}\n\n"
|
||||
f"Организация / Ведомство:\n{sig_info.get('org', 'Не указана')}\n\n"
|
||||
f"Серийный номер сертификата:\n{sig_info.get('serial', 'Не указан')}\n\n"
|
||||
f"Кем выдан сертификат:\n{sig_info.get('issuer', 'Удостоверяющий центр')}\n\n"
|
||||
f"Срок действия:\nС {sig_info.get('valid_from', '-')} по {sig_info.get('valid_to', '-')}\n\n"
|
||||
f"Дата обработки в системе Домовой 2.0:\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
||||
)
|
||||
|
||||
cert_page.insert_textbox(fitz.Rect(50, 110, 545, 450), details, fontsize=9.5, color=(0.15, 0.15, 0.15))
|
||||
|
||||
out_bytes = doc.tobytes()
|
||||
doc.close()
|
||||
return out_bytes
|
||||
except Exception as e:
|
||||
log.error(f"Failed to append certificate page: {e}")
|
||||
return pdf_bytes
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Signature Verifier ready.")
|
||||
Loading…
Reference in a new issue