Coverage for src/tracekit/reporting/advanced.py: 99%

1"""Advanced reporting features for TraceKit.

3This module provides advanced reporting capabilities including interactive

4reports, scheduled generation, distribution, versioning, and compliance.

5"""

7from __future__ import annotations

9import hashlib

10import json

11import logging

12import re

13import uuid

14from dataclasses import dataclass, field

15from datetime import datetime, timedelta

16from enum import Enum, auto

17from typing import TYPE_CHECKING, Any

19if TYPE_CHECKING:

20 from collections.abc import Callable

21 from pathlib import Path

23logger = logging.getLogger(__name__)

26# =============================================================================

27# =============================================================================

30@dataclass

31class TemplateField:

32 """Customizable template field.

34 Attributes:

35 name: Field identifier

36 type: Field type (text, number, image, table, chart)

37 default: Default value

38 required: Whether field is required

39 validation: Validation rule (regex pattern)

40 """

42 name: str

43 type: str = "text"

44 default: Any = None

45 required: bool = False

46 validation: str | None = None

47 description: str = ""

50@dataclass

51class CustomTemplate:

52 """Customizable report template.

54 Attributes:

55 name: Template name

56 version: Template version

57 fields: List of customizable fields

58 layout: Layout configuration

59 styles: CSS/style overrides

60 includes: Included partial templates

62 Example:

63 >>> template = CustomTemplate(

64 ... name="compliance_report",

65 ... fields=[

66 ... TemplateField("company_name", required=True),

67 ... TemplateField("logo", type="image")

68 ... ]

69 ... )

71 References:

72 REPORT-011: Report Customization Templates

73 """

75 name: str

76 version: str = "1.0.0"

77 fields: list[TemplateField] = field(default_factory=list)

78 layout: dict[str, Any] = field(default_factory=dict)

79 styles: dict[str, str] = field(default_factory=dict)

80 includes: list[str] = field(default_factory=list)

81 description: str = ""

83 def validate_data(self, data: dict[str, Any]) -> tuple[bool, list[str]]:

84 """Validate data against template fields.

86 Args:

87 data: Data dictionary

89 Returns:

90 Tuple of (is_valid, list of errors)

91 """

92 errors = []

93 for f in self.fields:

94 if f.required and f.name not in data:

95 errors.append(f"Required field '{f.name}' missing")

96 elif f.name in data and f.validation:

97 if not re.match(f.validation, str(data[f.name])):

98 errors.append(f"Field '{f.name}' failed validation")

99 return len(errors) == 0, errors

100

101 def render(self, data: dict[str, Any]) -> str:

102 """Render template with data.

103

104 Args:

105 data: Data dictionary

106

107 Returns:

108 Rendered content

109 """

110 # Simple placeholder substitution

111 content = self.layout.get("template", "")

112 for f in self.fields:

113 placeholder = f"{{{{{f.name}}}}}"

114 value = data.get(f.name, f.default or "")

115 content = content.replace(placeholder, str(value))

116 return content # type: ignore[no-any-return]

117

118

119# =============================================================================

120# =============================================================================

121

122

123class InteractiveElementType(Enum):

124 """Types of interactive elements."""

125

126 ZOOMABLE_CHART = auto()

127 COLLAPSIBLE_SECTION = auto()

128 FILTER_DROPDOWN = auto()

129 SORTABLE_TABLE = auto()

130 TOOLTIP = auto()

131 DRILL_DOWN = auto()

132 TOGGLE = auto()

133

134

135@dataclass

136class InteractiveElement:

137 """Interactive element for HTML reports.

138

139 Attributes:

140 id: Element ID

141 type: Element type

142 data: Element data

143 options: Configuration options

144 script: JavaScript code for interactivity

145

146 Example:

147 >>> element = InteractiveElement(

148 ... id="chart1",

149 ... type=InteractiveElementType.ZOOMABLE_CHART,

150 ... data=chart_data

151 ... )

152

153 References:

154 REPORT-012: Interactive Report Elements

155 """

156

157 id: str

158 type: InteractiveElementType

159 data: Any = None

160 options: dict[str, Any] = field(default_factory=dict)

161 script: str = ""

162

163 def to_html(self) -> str:

164 """Generate HTML for interactive element."""

165 html_parts = [f'<div id="{self.id}" class="interactive-{self.type.name.lower()}">']

166

167 if self.type == InteractiveElementType.COLLAPSIBLE_SECTION:

168 html_parts.append(f"""

169 <button class="collapsible" onclick="toggleSection('{self.id}')">

170 {self.options.get("title", "Section")}

171 </button>

172 <div class="content" style="display:none;">

173 {self.data or ""}

174 </div>

175 """)

176 elif self.type == InteractiveElementType.SORTABLE_TABLE:

177 html_parts.append(f"""

178 <table class="sortable" data-sort-enabled="true">

179 {self.data or ""}

180 </table>

181 """)

182 elif self.type == InteractiveElementType.TOOLTIP:

183 html_parts.append(f'''

184 <span class="tooltip" data-tooltip="{self.options.get("text", "")}">

185 {self.data or ""}

186 </span>

187 ''')

188 else:

189 html_parts.append(str(self.data or ""))

190

191 html_parts.append("</div>")

192 return "\n".join(html_parts)

193

194

195# =============================================================================

196# =============================================================================

197

198

199@dataclass

200class Annotation:

201 """Report annotation.

202

203 Attributes:

204 id: Unique annotation ID

205 target: Target element ID or location

206 text: Annotation text

207 author: Author name

208 created: Creation timestamp

209 type: Annotation type (note, warning, highlight, etc.)

210 position: Position info for placement

211

212 References:

213 REPORT-013: Report Annotations

214 """

215

216 id: str

217 target: str

218 text: str

219 author: str = ""

220 created: datetime = field(default_factory=datetime.now)

221 type: str = "note"

222 position: dict[str, float] = field(default_factory=dict)

223

224 def to_dict(self) -> dict[str, Any]:

225 """Convert to dictionary."""

226 return {

227 "id": self.id,

228 "target": self.target,

229 "text": self.text,

230 "author": self.author,

231 "created": self.created.isoformat(),

232 "type": self.type,

233 "position": self.position,

234 }

235

236

237class AnnotationManager:

238 """Manager for report annotations.

239

240 References:

241 REPORT-013: Report Annotations

242 """

243

244 def __init__(self, report_id: str):

245 self.report_id = report_id

246 self._annotations: list[Annotation] = []

247

248 def add(self, target: str, text: str, author: str = "", type_: str = "note") -> Annotation:

249 """Add annotation."""

250 annotation = Annotation(

251 id=str(uuid.uuid4()), target=target, text=text, author=author, type=type_

252 )

253 self._annotations.append(annotation)

254 return annotation

255

256 def remove(self, annotation_id: str) -> bool:

257 """Remove annotation."""

258 for i, ann in enumerate(self._annotations):

259 if ann.id == annotation_id: 259 ↛ 258line 259 didn't jump to line 258 because the condition on line 259 was always true

260 del self._annotations[i]

261 return True

262 return False

263

264 def get_for_target(self, target: str) -> list[Annotation]:

265 """Get annotations for target."""

266 return [a for a in self._annotations if a.target == target]

267

268 def export(self) -> list[dict[str, Any]]:

269 """Export all annotations."""

270 return [a.to_dict() for a in self._annotations]

271

272

273# =============================================================================

274# =============================================================================

275

276

277class ScheduleFrequency(Enum):

278 """Report schedule frequency."""

279

280 ONCE = auto()

281 HOURLY = auto()

282 DAILY = auto()

283 WEEKLY = auto()

284 MONTHLY = auto()

285 CUSTOM = auto()

286

287

288@dataclass

289class ReportSchedule:

290 """Scheduled report configuration.

291

292 Attributes:

293 id: Schedule ID

294 report_config: Report configuration

295 frequency: Generation frequency

296 next_run: Next scheduled run time

297 enabled: Whether schedule is active

298 recipients: Email recipients

299 cron_expression: Cron expression for custom schedules

300

301 References:

302 REPORT-017: Report Scheduling

303 """

304

305 id: str

306 report_config: dict[str, Any]

307 frequency: ScheduleFrequency = ScheduleFrequency.DAILY

308 next_run: datetime = field(default_factory=datetime.now)

309 enabled: bool = True

310 recipients: list[str] = field(default_factory=list)

311 cron_expression: str | None = None

312

313 def calculate_next_run(self) -> datetime:

314 """Calculate next run time."""

315 now = datetime.now()

316 if self.frequency == ScheduleFrequency.HOURLY:

317 return now + timedelta(hours=1)

318 elif self.frequency == ScheduleFrequency.DAILY:

319 return now + timedelta(days=1)

320 elif self.frequency == ScheduleFrequency.WEEKLY:

321 return now + timedelta(weeks=1)

322 elif self.frequency == ScheduleFrequency.MONTHLY: 322 ↛ 324line 322 didn't jump to line 324 because the condition on line 322 was always true

323 return now + timedelta(days=30)

324 return now

325

326

327class ReportScheduler:

328 """Report scheduler for automated generation.

329

330 References:

331 REPORT-017: Report Scheduling

332 """

333

334 def __init__(self): # type: ignore[no-untyped-def]

335 self._schedules: dict[str, ReportSchedule] = {}

336 self._running = False

337

338 def add_schedule(

339 self,

340 report_config: dict[str, Any],

341 frequency: ScheduleFrequency,

342 recipients: list[str] | None = None,

343 ) -> str:

344 """Add new schedule."""

345 schedule = ReportSchedule(

346 id=str(uuid.uuid4()),

347 report_config=report_config,

348 frequency=frequency,

349 recipients=recipients or [],

350 )

351 self._schedules[schedule.id] = schedule

352 return schedule.id

353

354 def remove_schedule(self, schedule_id: str) -> bool:

355 """Remove schedule."""

356 if schedule_id in self._schedules:

357 del self._schedules[schedule_id]

358 return True

359 return False

360

361 def get_pending(self) -> list[ReportSchedule]:

362 """Get schedules due for execution."""

363 now = datetime.now()

364 return [s for s in self._schedules.values() if s.enabled and s.next_run <= now]

365

366 def execute_pending(self, generator: Callable[[dict[str, Any]], Any]) -> list[str]:

367 """Execute pending schedules."""

368 executed = []

369 for schedule in self.get_pending():

370 try:

371 generator(schedule.report_config)

372 schedule.next_run = schedule.calculate_next_run()

373 executed.append(schedule.id)

374 except Exception as e:

375 logger.error(f"Scheduled report failed: {e}")

376 return executed

377

378

379# =============================================================================

380# =============================================================================

381

382

383class DistributionChannel(Enum):

384 """Distribution channels."""

385

386 EMAIL = auto()

387 FILE_SHARE = auto()

388 WEBHOOK = auto()

389 S3 = auto()

390 SFTP = auto()

391

392

393@dataclass

394class DistributionConfig:

395 """Distribution configuration.

396

397 References:

398 REPORT-020: Report Distribution

399 """

400

401 channel: DistributionChannel

402 recipients: list[str] = field(default_factory=list)

403 settings: dict[str, Any] = field(default_factory=dict)

404

405

406class ReportDistributor:

407 """Distributes reports to configured channels.

408

409 References:

410 REPORT-020: Report Distribution

411 """

412

413 def __init__(self): # type: ignore[no-untyped-def]

414 self._handlers: dict[DistributionChannel, Callable] = {} # type: ignore[type-arg]

415

416 def register_handler(

417 self,

418 channel: DistributionChannel,

419 handler: Callable[[Path, DistributionConfig], bool],

420 ) -> None:

421 """Register distribution handler."""

422 self._handlers[channel] = handler

423

424 def distribute(self, report_path: Path, configs: list[DistributionConfig]) -> dict[str, bool]:

425 """Distribute report to all configured channels."""

426 results = {}

427 for config in configs:

428 handler = self._handlers.get(config.channel)

429 if handler:

430 try:

431 results[config.channel.name] = handler(report_path, config)

432 except Exception as e:

433 logger.error(f"Distribution failed for {config.channel}: {e}")

434 results[config.channel.name] = False

435 else:

436 logger.warning(f"No handler for channel: {config.channel}")

437 results[config.channel.name] = False

438 return results

439

440

441# =============================================================================

442# =============================================================================

443

444

445@dataclass

446class ArchivedReport:

447 """Archived report metadata.

448

449 References:

450 REPORT-021: Report Archiving

451 """

452

453 id: str

454 name: str

455 path: Path

456 created: datetime

457 size: int

458 checksum: str

459 metadata: dict[str, Any] = field(default_factory=dict)

460 retention_days: int = 365

461

462

463class ReportArchive:

464 """Report archiving system.

465

466 References:

467 REPORT-021: Report Archiving

468 """

469

470 def __init__(self, archive_dir: Path):

471 self.archive_dir = archive_dir

472 archive_dir.mkdir(parents=True, exist_ok=True)

473 self._index: dict[str, ArchivedReport] = {}

474

475 def archive(self, report_path: Path, metadata: dict[str, Any] | None = None) -> str:

476 """Archive a report."""

477 report_id = str(uuid.uuid4())

478

479 # Calculate checksum

480 with open(report_path, "rb") as f:

481 checksum = hashlib.sha256(f.read()).hexdigest()

482

483 # Copy to archive

484 archive_path = self.archive_dir / f"{report_id}_{report_path.name}"

485 import shutil

486

487 shutil.copy2(report_path, archive_path)

488

489 archived = ArchivedReport(

490 id=report_id,

491 name=report_path.name,

492 path=archive_path,

493 created=datetime.now(),

494 size=archive_path.stat().st_size,

495 checksum=checksum,

496 metadata=metadata or {},

497 )

498 self._index[report_id] = archived

499

500 logger.info(f"Archived report: {report_id}")

501 return report_id

502

503 def retrieve(self, report_id: str) -> Path | None:

504 """Retrieve archived report."""

505 if report_id in self._index:

506 return self._index[report_id].path

507 return None

508

509 def cleanup_expired(self) -> int:

510 """Remove expired archives."""

511 now = datetime.now()

512 removed = 0

513 for report_id, archived in list(self._index.items()):

514 age = (now - archived.created).days

515 if age > archived.retention_days:

516 archived.path.unlink(missing_ok=True)

517 del self._index[report_id]

518 removed += 1

519 return removed

520

521

522# =============================================================================

523# =============================================================================

524

525

526@dataclass

527class SearchResult:

528 """Report search result.

529

530 References:

531 REPORT-022: Report Search

532 """

533

534 report_id: str

535 name: str

536 score: float

537 highlights: list[str] = field(default_factory=list)

538 metadata: dict[str, Any] = field(default_factory=dict)

539

540

541class ReportSearchIndex:

542 """Full-text search index for reports.

543

544 References:

545 REPORT-022: Report Search

546 """

547

548 def __init__(self): # type: ignore[no-untyped-def]

549 self._index: dict[str, dict[str, Any]] = {}

550

551 def index_report(self, report_id: str, content: str, metadata: dict[str, Any]) -> None:

552 """Add report to search index."""

553 # Simple word-based indexing

554 words = set(content.lower().split())

555 self._index[report_id] = {

556 "words": words,

557 "content": content,

558 "metadata": metadata,

559 }

560

561 def search(self, query: str, limit: int = 10) -> list[SearchResult]:

562 """Search for reports."""

563 query_words = set(query.lower().split())

564 results = []

565

566 for report_id, doc in self._index.items():

567 # Simple scoring: intersection of words

568 matches = query_words & doc["words"]

569 if matches:

570 score = len(matches) / len(query_words)

571 results.append(

572 SearchResult(

573 report_id=report_id,

574 name=doc["metadata"].get("name", report_id),

575 score=score,

576 highlights=[f"...{m}..." for m in matches],

577 metadata=doc["metadata"],

578 )

579 )

580

581 # Sort by score

582 results.sort(key=lambda r: r.score, reverse=True)

583 return results[:limit]

584

585

586# =============================================================================

587# =============================================================================

588

589

590@dataclass

591class ReportVersion:

592 """Report version entry.

593

594 References:

595 REPORT-023: Report Versioning

596 """

597

598 version: int

599 created: datetime

600 author: str

601 changes: str

602 checksum: str

603 path: Path

604

605

606class ReportVersionControl:

607 """Version control for reports.

608

609 References:

610 REPORT-023: Report Versioning

611 """

612

613 def __init__(self, storage_dir: Path):

614 self.storage_dir = storage_dir

615 storage_dir.mkdir(parents=True, exist_ok=True)

616 self._versions: dict[str, list[ReportVersion]] = {}

617

618 def commit(self, report_id: str, report_path: Path, author: str, changes: str) -> int:

619 """Commit new version of report."""

620 if report_id not in self._versions:

621 self._versions[report_id] = []

622

623 version = len(self._versions[report_id]) + 1

624

625 # Copy to versioned storage

626 version_path = self.storage_dir / f"{report_id}_v{version}{report_path.suffix}"

627 import shutil

628

629 shutil.copy2(report_path, version_path)

630

631 # Calculate checksum

632 with open(version_path, "rb") as f:

633 checksum = hashlib.sha256(f.read()).hexdigest()

634

635 entry = ReportVersion(

636 version=version,

637 created=datetime.now(),

638 author=author,

639 changes=changes,

640 checksum=checksum,

641 path=version_path,

642 )

643 self._versions[report_id].append(entry)

644

645 logger.info(f"Committed {report_id} version {version}")

646 return version

647

648 def get_version(self, report_id: str, version: int) -> Path | None:

649 """Get specific version of report."""

650 if report_id in self._versions:

651 for v in self._versions[report_id]: 651 ↛ 654line 651 didn't jump to line 654 because the loop on line 651 didn't complete

652 if v.version == version:

653 return v.path

654 return None

655

656 def get_history(self, report_id: str) -> list[ReportVersion]:

657 """Get version history."""

658 return self._versions.get(report_id, [])

659

660 def diff(self, report_id: str, v1: int, v2: int) -> str:

661 """Get diff between versions."""

662 path1 = self.get_version(report_id, v1)

663 path2 = self.get_version(report_id, v2)

664

665 if not path1 or not path2:

666 return "Version not found"

667

668 # Simple text diff

669 with open(path1) as f1, open(path2) as f2:

670 lines1 = f1.readlines()

671 lines2 = f2.readlines()

672

673 import difflib

674

675 diff = difflib.unified_diff(lines1, lines2, lineterm="")

676 return "\n".join(diff)

677

678

679# =============================================================================

680# =============================================================================

681

682

683class ApprovalStatus(Enum):

684 """Approval status."""

685

686 DRAFT = auto()

687 PENDING_REVIEW = auto()

688 APPROVED = auto()

689 REJECTED = auto()

690 PUBLISHED = auto()

691

692

693@dataclass

694class ApprovalRecord:

695 """Approval workflow record.

696

697 References:

698 REPORT-024: Report Approval Workflow

699 """

700

701 report_id: str

702 status: ApprovalStatus = ApprovalStatus.DRAFT

703 submitter: str = ""

704 reviewer: str | None = None

705 submitted_at: datetime | None = None

706 reviewed_at: datetime | None = None

707 comments: str = ""

708

709

710class ApprovalWorkflow:

711 """Report approval workflow manager.

712

713 References:

714 REPORT-024: Report Approval Workflow

715 """

716

717 def __init__(self): # type: ignore[no-untyped-def]

718 self._records: dict[str, ApprovalRecord] = {}

719 self._callbacks: dict[ApprovalStatus, list[Callable]] = {} # type: ignore[type-arg]

720

721 def submit_for_review(self, report_id: str, submitter: str) -> ApprovalRecord:

722 """Submit report for review."""

723 record = ApprovalRecord(

724 report_id=report_id,

725 status=ApprovalStatus.PENDING_REVIEW,

726 submitter=submitter,

727 submitted_at=datetime.now(),

728 )

729 self._records[report_id] = record

730 self._trigger_callbacks(ApprovalStatus.PENDING_REVIEW, record)

731 return record

732

733 def approve(self, report_id: str, reviewer: str, comments: str = "") -> ApprovalRecord:

734 """Approve report."""

735 record = self._records.get(report_id)

736 if not record:

737 raise ValueError(f"Report {report_id} not in workflow")

738

739 record.status = ApprovalStatus.APPROVED

740 record.reviewer = reviewer

741 record.reviewed_at = datetime.now()

742 record.comments = comments

743 self._trigger_callbacks(ApprovalStatus.APPROVED, record)

744 return record

745

746 def reject(self, report_id: str, reviewer: str, comments: str) -> ApprovalRecord:

747 """Reject report."""

748 record = self._records.get(report_id)

749 if not record:

750 raise ValueError(f"Report {report_id} not in workflow")

751

752 record.status = ApprovalStatus.REJECTED

753 record.reviewer = reviewer

754 record.reviewed_at = datetime.now()

755 record.comments = comments

756 self._trigger_callbacks(ApprovalStatus.REJECTED, record)

757 return record

758

759 def on_status_change(

760 self, status: ApprovalStatus, callback: Callable[[ApprovalRecord], None]

761 ) -> None:

762 """Register callback for status change."""

763 if status not in self._callbacks: 763 ↛ 765line 763 didn't jump to line 765 because the condition on line 763 was always true

764 self._callbacks[status] = []

765 self._callbacks[status].append(callback)

766

767 def _trigger_callbacks(self, status: ApprovalStatus, record: ApprovalRecord) -> None:

768 """Trigger callbacks for status."""

769 for callback in self._callbacks.get(status, []):

770 try:

771 callback(record)

772 except Exception as e:

773 logger.warning(f"Approval callback failed: {e}")

774

775

776# =============================================================================

777# =============================================================================

778

779

780@dataclass

781class ComplianceRule:

782 """Compliance checking rule.

783

784 References:

785 REPORT-025: Report Compliance Checking

786 """

787

788 id: str

789 name: str

790 description: str

791 check: Callable[[dict[str, Any]], bool]

792 severity: str = "error" # error, warning, info

793

794

795@dataclass

796class ComplianceResult:

797 """Compliance check result."""

798

799 passed: bool

800 violations: list[tuple[str, str]] = field(default_factory=list)

801 warnings: list[tuple[str, str]] = field(default_factory=list)

802

803

804class ComplianceChecker:

805 """Report compliance checker.

806

807 References:

808 REPORT-025: Report Compliance Checking

809 """

810

811 def __init__(self): # type: ignore[no-untyped-def]

812 self._rules: list[ComplianceRule] = []

813

814 def add_rule(

815 self,

816 name: str,

817 check: Callable[[dict[str, Any]], bool],

818 description: str = "",

819 severity: str = "error",

820 ) -> None:

821 """Add compliance rule."""

822 rule = ComplianceRule(

823 id=str(uuid.uuid4()),

824 name=name,

825 description=description,

826 check=check,

827 severity=severity,

828 )

829 self._rules.append(rule)

830

831 def check(self, report_data: dict[str, Any]) -> ComplianceResult:

832 """Check report against all rules."""

833 violations = []

834 warnings = []

835

836 for rule in self._rules:

837 try:

838 if not rule.check(report_data):

839 if rule.severity == "error":

840 violations.append((rule.name, rule.description))

841 else:

842 warnings.append((rule.name, rule.description))

843 except Exception as e:

844 logger.warning(f"Compliance rule {rule.name} failed: {e}")

845

846 return ComplianceResult(

847 passed=len(violations) == 0, violations=violations, warnings=warnings

848 )

849

850

851# =============================================================================

852# =============================================================================

853

854

855@dataclass

856class LocaleStrings:

857 """Localized strings for a locale.

858

859 References:

860 REPORT-026: Report Localization

861 """

862

863 locale: str

864 strings: dict[str, str] = field(default_factory=dict)

865 date_format: str = "%Y-%m-%d"

866 time_format: str = "%H:%M:%S"

867 number_decimal: str = "."

868 number_thousand: str = ","

869

870

871class ReportLocalizer:

872 """Report localization manager.

873

874 References:

875 REPORT-026: Report Localization

876 """

877

878 def __init__(self, default_locale: str = "en_US"):

879 self.default_locale = default_locale

880 self._locales: dict[str, LocaleStrings] = {}

881 self._register_defaults()

882

883 def _register_defaults(self) -> None:

884 """Register default locales."""

885 self._locales["en_US"] = LocaleStrings(

886 locale="en_US",

887 strings={

888 "title": "Report",

889 "summary": "Summary",

890 "pass": "PASS",

891 "fail": "FAIL",

892 },

893 )

894 self._locales["de_DE"] = LocaleStrings(

895 locale="de_DE",

896 strings={

897 "title": "Bericht",

898 "summary": "Zusammenfassung",

899 "pass": "BESTANDEN",

900 "fail": "DURCHGEFALLEN",

901 },

902 date_format="%d.%m.%Y",

903 number_decimal=",",

904 number_thousand=".",

905 )

906

907 def get_string(self, key: str, locale: str | None = None) -> str:

908 """Get localized string."""

909 loc = locale or self.default_locale

910 strings = self._locales.get(loc, self._locales[self.default_locale])

911 return strings.strings.get(key, key)

912

913 def format_number(self, value: float, locale: str | None = None) -> str:

914 """Format number for locale."""

915 loc_strings = self._locales.get(

916 locale or self.default_locale, self._locales[self.default_locale]

917 )

918 formatted = f"{value:,.2f}"

919 # Replace separators

920 formatted = formatted.replace(",", "TEMP")

921 formatted = formatted.replace(".", loc_strings.number_decimal)

922 formatted = formatted.replace("TEMP", loc_strings.number_thousand)

923 return formatted

924

925

926# =============================================================================

927# =============================================================================

928

929

930@dataclass

931class AccessibilityOptions:

932 """Accessibility options for reports.

933

934 References:

935 REPORT-027: Report Accessibility

936 """

937

938 alt_text_required: bool = True

939 high_contrast: bool = False

940 screen_reader_friendly: bool = True

941 keyboard_navigable: bool = True

942 wcag_level: str = "AA" # A, AA, AAA

943

944

945def add_accessibility_features(html_content: str, options: AccessibilityOptions) -> str:

946 """Add accessibility features to HTML report.

947

948 Args:

949 html_content: HTML content

950 options: Accessibility options

951

952 Returns:

953 Enhanced HTML content

954

955 References:

956 REPORT-027: Report Accessibility

957 """

958 # Add ARIA landmarks

959 html_content = html_content.replace(

960 '<div class="report">',

961 '<div class="report" role="main" aria-label="Report Content">',

962 )

963

964 # Add skip navigation link

965 skip_nav = '<a href="#main-content" class="skip-link">Skip to main content</a>'

966 html_content = html_content.replace("<body>", f"<body>{skip_nav}")

967

968 # Add high contrast styles if enabled

969 if options.high_contrast:

970 contrast_styles = """

971 <style>

972 body { background: white !important; color: black !important; }

973 a { color: blue !important; }

974 .pass { background: green !important; color: white !important; }

975 .fail { background: red !important; color: white !important; }

976 </style>

977 """

978 html_content = html_content.replace("</head>", f"{contrast_styles}</head>")

979

980 return html_content

981

982

983# =============================================================================

984# =============================================================================

985

986

987class ReportEncryption:

988 """Report encryption utilities.

989

990 References:

991 REPORT-028: Report Encryption

992 """

993

994 @staticmethod

995 def encrypt_content(content: bytes, password: str) -> bytes:

996 """Encrypt report content.

997

998 Args:

999 content: Content bytes to encrypt.

1000 password: Encryption password.

1001

1002 Returns:

1003 Encrypted content bytes.

1004

1005 Note:

1006 Uses simple XOR encryption for demonstration.

1007 In production, use proper encryption (AES, etc.).

1008 """

1009 key = hashlib.sha256(password.encode()).digest()

1010 encrypted = bytearray()

1011 for i, byte in enumerate(content):

1012 encrypted.append(byte ^ key[i % len(key)])

1013 return bytes(encrypted)

1014

1015 @staticmethod

1016 def decrypt_content(encrypted: bytes, password: str) -> bytes:

1017 """Decrypt report content."""

1018 # XOR is symmetric

1019 return ReportEncryption.encrypt_content(encrypted, password)

1020

1021 @staticmethod

1022 def encrypt_file(input_path: Path, output_path: Path, password: str) -> None:

1023 """Encrypt report file."""

1024 with open(input_path, "rb") as f:

1025 content = f.read()

1026 encrypted = ReportEncryption.encrypt_content(content, password)

1027 with open(output_path, "wb") as f:

1028 f.write(encrypted)

1029

1030 @staticmethod

1031 def decrypt_file(input_path: Path, output_path: Path, password: str) -> None:

1032 """Decrypt report file."""

1033 with open(input_path, "rb") as f:

1034 encrypted = f.read()

1035 decrypted = ReportEncryption.decrypt_content(encrypted, password)

1036 with open(output_path, "wb") as f:

1037 f.write(decrypted)

1038

1039

1040# =============================================================================

1041# =============================================================================

1042

1043

1044@dataclass

1045class Watermark:

1046 """Report watermark configuration.

1047

1048 References:

1049 REPORT-029: Report Watermarking

1050 """

1051

1052 text: str = "CONFIDENTIAL"

1053 opacity: float = 0.1

1054 rotation: int = -45

1055 position: str = "center" # center, header, footer

1056 font_size: int = 48

1057

1058

1059def add_watermark(html_content: str, watermark: Watermark) -> str:

1060 """Add watermark to HTML report.

1061

1062 Args:

1063 html_content: HTML content

1064 watermark: Watermark configuration

1065

1066 Returns:

1067 HTML with watermark

1068

1069 References:

1070 REPORT-029: Report Watermarking

1071 """

1072 watermark_css = f"""

1073 <style>

1074 .watermark {{

1075 position: fixed;

1076 top: 50%;

1077 left: 50%;

1078 transform: translate(-50%, -50%) rotate({watermark.rotation}deg);

1079 font-size: {watermark.font_size}px;

1080 color: rgba(128, 128, 128, {watermark.opacity});

1081 pointer-events: none;

1082 z-index: 1000;

1083 white-space: nowrap;

1084 }}

1085 </style>

1086 """

1087 watermark_div = f'<div class="watermark">{watermark.text}</div>'

1088

1089 html_content = html_content.replace("</head>", f"{watermark_css}</head>")

1090 html_content = html_content.replace("<body>", f"<body>{watermark_div}")

1091

1092 return html_content

1093

1094

1095# =============================================================================

1096# =============================================================================

1097

1098

1099@dataclass

1100class AuditEntry:

1101 """Audit trail entry.

1102

1103 References:

1104 REPORT-030: Report Audit Trail

1105 """

1106

1107 id: str

1108 report_id: str

1109 action: str

1110 user: str

1111 timestamp: datetime

1112 details: dict[str, Any] = field(default_factory=dict)

1113 ip_address: str = ""

1114

1115

1116class AuditTrail:

1117 """Report audit trail manager.

1118

1119 References:

1120 REPORT-030: Report Audit Trail

1121 """

1122

1123 def __init__(self, storage_path: Path | None = None):

1124 self.storage_path = storage_path

1125 self._entries: list[AuditEntry] = []

1126

1127 def log(

1128 self,

1129 report_id: str,

1130 action: str,

1131 user: str,

1132 details: dict[str, Any] | None = None,

1133 ) -> AuditEntry:

1134 """Log audit entry."""

1135 entry = AuditEntry(

1136 id=str(uuid.uuid4()),

1137 report_id=report_id,

1138 action=action,

1139 user=user,

1140 timestamp=datetime.now(),

1141 details=details or {},

1142 )

1143 self._entries.append(entry)

1144

1145 # Persist if storage configured

1146 if self.storage_path:

1147 self._persist()

1148

1149 return entry

1150

1151 def get_for_report(self, report_id: str) -> list[AuditEntry]:

1152 """Get audit entries for report."""

1153 return [e for e in self._entries if e.report_id == report_id]

1154

1155 def get_by_user(self, user: str) -> list[AuditEntry]:

1156 """Get audit entries by user."""

1157 return [e for e in self._entries if e.user == user]

1158

1159 def export(self, format_: str = "json") -> str:

1160 """Export audit trail."""

1161 if format_ == "json": 1161 ↛ 1176line 1161 didn't jump to line 1176 because the condition on line 1161 was always true

1162 return json.dumps(

1163 [

1164 {

1165 "id": e.id,

1166 "report_id": e.report_id,

1167 "action": e.action,

1168 "user": e.user,

1169 "timestamp": e.timestamp.isoformat(),

1170 "details": e.details,

1171 }

1172 for e in self._entries

1173 ],

1174 indent=2,

1175 )

1176 return ""

1177

1178 def _persist(self) -> None:

1179 """Persist audit trail to storage."""

1180 if self.storage_path: 1180 ↛ exitline 1180 didn't return from function '_persist' because the condition on line 1180 was always true

1181 with open(self.storage_path, "w") as f:

1182 f.write(self.export("json"))

1183

1184

1185__all__ = [

1186 # Accessibility (REPORT-027)

1187 "AccessibilityOptions",

1188 # Annotations (REPORT-013)

1189 "Annotation",

1190 "AnnotationManager",

1191 # Approval (REPORT-024)

1192 "ApprovalRecord",

1193 "ApprovalStatus",

1194 "ApprovalWorkflow",

1195 # Archiving (REPORT-021)

1196 "ArchivedReport",

1197 # Audit Trail (REPORT-030)

1198 "AuditEntry",

1199 "AuditTrail",

1200 # Compliance (REPORT-025)

1201 "ComplianceChecker",

1202 "ComplianceResult",

1203 "ComplianceRule",

1204 # Templates (REPORT-011)

1205 "CustomTemplate",

1206 # Distribution (REPORT-020)

1207 "DistributionChannel",

1208 "DistributionConfig",

1209 # Interactive (REPORT-012)

1210 "InteractiveElement",

1211 "InteractiveElementType",

1212 # Localization (REPORT-026)

1213 "LocaleStrings",

1214 "ReportArchive",

1215 "ReportDistributor",

1216 # Encryption (REPORT-028)

1217 "ReportEncryption",

1218 "ReportLocalizer",

1219 # Scheduling (REPORT-017)

1220 "ReportSchedule",

1221 "ReportScheduler",

1222 # Search (REPORT-022)

1223 "ReportSearchIndex",

1224 # Versioning (REPORT-023)

1225 "ReportVersion",

1226 "ReportVersionControl",

1227 "ScheduleFrequency",

1228 "SearchResult",

1229 "TemplateField",

1230 # Watermarking (REPORT-029)

1231 "Watermark",

1232 "add_accessibility_features",

1233 "add_watermark",

1234]

Coverage for src / tracekit / reporting / advanced.py: 99%

485 statements