knowledge-governance/scripts/qa_ops_directory_sync.py

"""Read-only bridge to the existing weekly-report contacts fetchers.

The bridge does not execute the weekly report, export spreadsheets, write its
configuration, or infer company/department names. All new data stays in a private
caller-selected JSON file for the governance importer.
"""
import argparse
from datetime import datetime, timezone
import importlib
import json
import logging
import os
from pathlib import Path
import sys
import types
from urllib.request import HTTPRedirectHandler, ProxyHandler, build_opener

sys.dont_write_bytecode = True


class DirectorySyncError(RuntimeError):
    pass


def _validate_directory(departments, people):
    if not departments or not people or len(departments) > 20000 or len(people) > 100000:
        raise DirectorySyncError('directory_integrity_failed')
    for identity, department in departments.items():
        if identity <= 0 or not department.get('name'):
            raise DirectorySyncError('directory_integrity_failed')
        seen = set()
        current = identity
        while current:
            if current in seen or current not in departments or len(seen) > 100:
                raise DirectorySyncError('directory_integrity_failed')
            seen.add(current)
            current = int(departments[current].get('parent_id') or 0)
    for identity, person in people.items():
        name = person.get('name')
        if name is not None and (not isinstance(name, str) or len(name.encode('utf-8')) > 1000 or '\x00' in name):
            raise DirectorySyncError('directory_integrity_failed')
        if not identity or str(person.get('user_id') or '') != str(identity):
            raise DirectorySyncError('directory_integrity_failed')
        if not person.get('department_ids') or any(int(d) not in departments for d in person['department_ids']):
            raise DirectorySyncError('directory_integrity_failed')


def collect_directory(client, fetch_departments, fetch_users, get_detail, logger):
    if client.config.dry_run:
        raise DirectorySyncError('directory_dry_run_rejected')
    departments, denied, failed = fetch_departments(client, logger, None, None)
    if denied or failed or not departments:
        raise DirectorySyncError('directory_department_fetch_incomplete')
    departments = dict(departments)
    # Legacy full traversal returns children but omits the starting root. Read
    # actual missing ancestors; never manufacture a root or discard parent IDs.
    for _ in range(100):
        missing = sorted({int(d.get('parent_id') or 0) for d in departments.values()} - set(departments) - {0})
        if not missing:
            break
        for identity in missing:
            detail, denied, failed = get_detail(client, logger, identity)
            if denied or failed or not detail or int(detail.get('id') or 0) != identity or not detail.get('name'):
                raise DirectorySyncError('directory_ancestor_unavailable')
            departments[identity] = detail
    else:
        raise DirectorySyncError('directory_integrity_failed')
    people, denied, failed = fetch_users(client, logger, sorted(departments), enable_user_detail=False)
    if denied or failed:
        raise DirectorySyncError('directory_people_fetch_incomplete')
    _validate_directory(departments, people)
    return {
        'source': 'weekly_report_live',
        'synced_at': datetime.now(timezone.utc).isoformat(),
        'complete': True,
        'directory': {
            'departments': [
                {'id': str(identity), 'parent_id': str(int(d.get('parent_id') or 0)) if d.get('parent_id') else '', 'name': str(d['name'])}
                for identity, d in sorted(departments.items())
            ],
            # List membership provides stable staff IDs; do not reuse the old
            # guessed primary department, employment policy or openId fallback.
            'people': [
                {'id': str(identity), **({'name': p['name']} if p.get('name') else {}), 'union_id': str((p.get('raw_payload') or {}).get('unionid') or (p.get('raw_payload') or {}).get('unionId') or ''), 'department_ids': [str(d) for d in sorted(p['department_ids'])]}
                for identity, p in sorted(people.items())
            ],
        },
    }


class _NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise DirectorySyncError('directory_redirect_rejected')


def load_existing_runtime(root):
    root = Path(root)
    if not root.is_absolute() or not (root / 'scripts/dingtalk_sync/client.py').is_file():
        raise DirectorySyncError('directory_configuration_invalid')
    secret = root / '.env'
    info = secret.lstat()
    if not secret.is_file() or secret.is_symlink() or info.st_mode & 0o077:
        raise DirectorySyncError('directory_configuration_invalid')
    package_name = '_qaops_weekly_directory'
    package = types.ModuleType(package_name)
    package.__path__ = [str(root / 'scripts/dingtalk_sync')]
    sys.modules[package_name] = package
    client_module = importlib.import_module(package_name + '.client')
    dept_module = importlib.import_module(package_name + '.fetch_departments')
    people_module = importlib.import_module(package_name + '.fetch_users')
    config = client_module.load_sync_config(str(secret), str(root / 'config/dingtalk_sync.yaml'))
    if config.api_base_url.rstrip('/') != 'https://oapi.dingtalk.com' or config.dry_run or config.api_mode != 'topapi_v2':
        raise DirectorySyncError('directory_configuration_invalid')
    if not (1 <= config.timeout_seconds <= 30 and 0 <= config.max_retries <= 3 and 1 <= config.page_size <= 100):
        raise DirectorySyncError('directory_configuration_invalid')
    # Keep the old transport inside this private module namespace. Suppress its
    # potentially sensitive exception logging and refuse redirects/proxies.
    logger = logging.getLogger('qa_ops_directory_private')
    logger.handlers = [logging.NullHandler()]
    logger.propagate = False
    opener = build_opener(ProxyHandler({}), _NoRedirect())
    setattr(client_module, 'urlopen', opener.open)

    class BoundedClient(client_module.DingtalkClient):
        def __init__(self):
            super().__init__(config, logger=logger)
            self.requests = 0
            self.pages = set()

        def request_json(self, method, path, params=None, json_body=None, retry_on_business_error=True):
            self.requests += 1
            if self.requests > 10000:
                raise DirectorySyncError('directory_request_limit')
            if path.endswith('/listsub') or path.endswith('/user/list'):
                key = (path, json.dumps(json_body, sort_keys=True))
                if key in self.pages:
                    raise DirectorySyncError('directory_pagination_repeated')
                self.pages.add(key)
            return super().request_json(method, path, params, json_body, retry_on_business_error)

    return BoundedClient(), dept_module.fetch_all_departments, people_module.fetch_users, dept_module._get_department_detail, logger


def main():
    parser = argparse.ArgumentParser(description='Fetch a complete read-only directory into private governance JSON; no weekly-report writes')
    parser.add_argument('--runtime-root', required=True)
    parser.add_argument('--output', required=True)
    parser.add_argument('--fetch-live-directory', action='store_true', required=True)
    args = parser.parse_args()
    output = Path(args.output)
    if not output.is_absolute() or output.exists():
        raise DirectorySyncError('directory_output_must_be_new_absolute_path')
    client, departments, people, detail, logger = load_existing_runtime(args.runtime_root)
    print(json.dumps({'stage': 'fetching_live_directory', 'weekly_report_outputs_modified': False}), flush=True)
    result = collect_directory(client, departments, people, detail, logger)
    raw = json.dumps(result, ensure_ascii=False).encode('utf-8')
    if len(raw) > 16 << 20:
        raise DirectorySyncError('directory_output_limit')
    fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(fd, 'wb') as stream:
            stream.write(raw)
            stream.flush()
            os.fsync(stream.fileno())
    except BaseException:
        output.unlink(missing_ok=True)
        raise
    print(json.dumps({'stage': 'complete', 'departments': len(result['directory']['departments']), 'people': len(result['directory']['people']), 'synced_at': result['synced_at'], 'requests': client.requests, 'complete': True, 'employee_identities_logged': False, 'weekly_report_outputs_modified': False}), flush=True)


if __name__ == '__main__':
    try:
        main()
    except Exception as failure:
        code = str(failure) if isinstance(failure, DirectorySyncError) else 'directory_sync_failed'
        print(json.dumps({'stage': 'failed', 'code': code}), file=sys.stderr)
        raise SystemExit(1)