불완전한 자료를 한 줄로 정리하기 전에 어떤 행을 같은 글로 묶었는지 읽을 수 있어야 했다. 완전 식별 그룹과 제목 검토 후보를 따로 돌려주는 작은 함수를 공개한다.
행을 묶기 전에 입력 계약을 작게 정했다
이 예제의 입력은 여섯 필드뿐이다. row_id는 이번에 전달한 행의 별칭이고, source는 web 또는 mobile이다. namespace는 자료 집합의 별칭, owner는 게시자 별칭, record는 글 번호 문자열, title은 비교할 합성 제목이다. 실제 주소·본문·계정 자료를 받는 수집기는 포함하지 않았다. 원본 자료를 자동 익명화하는 도구도 아니므로 공개 예제에는 만든 별칭을 넣어야 한다.
한 호출은 64행까지 받고 제목은 240자로 제한했다. row_id 중복이나 모르는 필드가 있으면 전체 호출을 거절한다. 거절된 입력을 보존하는 책임은 호출 측에 있다. 행 번호를 바꿔 중복을 숨기는 대신, 다른 관측 두 개에 실제로 다른 행 별칭이 있는지 먼저 정해야 한다.
선언한 식별자와 미확인 행은 다르게 묶인다
namespace·소문자로 정리한 owner·record가 모두 있을 때만 선언된 식별 그룹을 만든다. owner는 예제에서 허용한 ASCII 별칭에 한해서 대소문자를 같게 본다. namespace는 대소문자를 구분하고 record의 앞자리 0도 지우지 않는다. 같은 숫자처럼 보여도 101과 0101을 같은 글로 바꾸지 않는 정책이다. 실제 서비스에 적용할 때 이 규칙을 먼저 검토해야 한다.
owner 또는 record가 없으면 그 행마다 독립적인 unresolved 그룹을 만든다. 제목이 같아도 다른 그룹에 자동으로 붙이지 않는다. 같은 선언 식별자의 여러 행은 하나의 그룹 아래 observations에 모두 남는다. 두 채널에서 받은 제목이 다르거나 같은 채널이 반복돼도 어느 행 하나를 대표로 골라 버리지 않는다.
제목 비교는 별도의 검토 후보를 만든다
제목 보조 키는 영문 소문자·숫자·한글 음절만 남긴다. A-B와 AB가 같아질 수 있고 이모지만 있는 제목은 빈 값이 된다. 의도적으로 정보가 줄어드는 예시다. namespace와 owner, 비어 있지 않은 제목 키가 같으면서 서로 다른 그룹에 속한 행들을 검토 후보로 돌려준다. 빈 키나 owner가 없는 행에는 제목 후보를 만들지 않는다.
후보가 없다는 것도 새로운 글이라는 증명은 아니다. 이 정규화가 다루지 않는 언어나 다른 제목을 쓴 같은 글은 후보에서 빠질 수 있다. 두 완전 식별자가 같은 제목을 쓰면 title_collision으로 남기고, 미확인 그룹이 끼어 있으면 unresolved_title_candidate로 남긴다. 어느 경우에도 후보를 정본 선택이나 병합 승인으로 바꾸지 않는다.
코드와 구현 살펴보기
독립 합성 전체 구현
"""Independent synthetic grouping; no source parsing, authorization or writes."""
from copy import deepcopy
import re
FIELDS = {'row_id', 'source', 'namespace', 'owner', 'record', 'title'}
ALIAS = re.compile(r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,47}\Z')
def title_key(title):
# Lossy and deliberately limited. This is a review hint, never an identity.
return re.sub(r'[^0-9a-z가-힣]+', '', title.lower())
def project(rows):
if not isinstance(rows, list) or len(rows) > 64:
raise ValueError('Expected a list of at most 64 synthetic rows')
clean, seen = [], set()
for original in rows:
if not isinstance(original, dict) or set(original) != FIELDS:
raise ValueError('Unexpected or missing fields')
row = deepcopy(original)
for field in ['row_id', 'namespace']:
if not isinstance(row[field], str) or not ALIAS.fullmatch(row[field]):
raise ValueError('Invalid alias')
if row['row_id'] in seen:
raise ValueError('Repeated row_id; preserve the evidence outside this call')
seen.add(row['row_id'])
if row['source'] not in ('web', 'mobile'):
raise ValueError('Unknown source')
if not isinstance(row['title'], str) or len(row['title']) > 240:
raise ValueError('Invalid title')
owner, record = row['owner'], row['record']
if owner not in (None, '') and (not isinstance(owner, str) or not ALIAS.fullmatch(owner)):
raise ValueError('Invalid owner alias')
if record not in (None, '') and (not isinstance(record, str) or not re.fullmatch(r'[0-9]{1,24}', record)):
raise ValueError('Record must be a decimal string or missing')
clean.append(row)
groups, buckets, row_group = {}, {}, {}
for row in sorted(clean, key=lambda x: x['row_id']):
known = bool(row['owner'] and row['record'])
identity = (row['namespace'], row['owner'].lower(), row['record']) if known else None
group_id = 'known:' + ':'.join(identity) if known else 'unresolved:' + row['row_id']
group = groups.setdefault(group_id, {
'group_id': group_id,
'basis': 'declared_identifier' if known else 'unresolved',
'declared_identity': list(identity) if known else None,
'member_row_ids': [], 'observations': []})
group['member_row_ids'].append(row['row_id'])
group['observations'].append(row)
row_group[row['row_id']] = group_id
key = title_key(row['title'])
if row['owner'] and key:
bucket = (row['namespace'], row['owner'].lower(), key)
buckets.setdefault(bucket, []).append(row['row_id'])
reviews = []
for (namespace, owner, key), members in sorted(buckets.items()):
ids = sorted({row_group[x] for x in members})
if len(ids) < 2:
continue
reviews.append({'namespace': namespace, 'owner_folded': owner,
'title_key': key, 'member_row_ids': sorted(members),
'group_ids': ids,
'reason': 'title_collision' if all(x.startswith('known:') for x in ids)
else 'unresolved_title_candidate'})
ordered = [groups[k] for k in sorted(groups)]
return {'groups': ordered, 'review_candidates': reviews,
'input_rows': len(clean),
'known_groups': sum(g['basis'] == 'declared_identifier' for g in ordered),
'unresolved_groups': sum(g['basis'] == 'unresolved' for g in ordered),
'identity_authenticated': False,
'normalized_title_proves_identity': False,
'merge_authorized': False, 'performs_external_write': False}
한 행도 사라지지 않는지 출력으로 확인한다
input_rows와 모든 그룹의 member_row_ids를 함께 읽으면 받은 행이 어디에 남았는지 확인할 수 있다. observations에는 허용된 여섯 필드가 그대로 복사된다. 입력 객체와 출력 객체를 분리해, 호출 후 화면 쪽에서 제목을 고쳐도 원래 입력은 바뀌지 않게 했다. 정상 입력을 받은 경우의 보존이며 거절한 입력까지 반환한다는 뜻은 아니다.
입력 순서를 바꿔도 같은 결과가 나오도록 행·그룹·후보를 정렬했다. 다만 row_id 자체를 바꾸면 같은 결과를 약속하지 않는다. 같은 글의 관측 시각이나 최신성은 입력에 없으므로 이 코드로 어느 제목이 최신인지 고를 수도 없다. 실제 적용에서는 원장을 따로 보존하고 어떤 자료 집합을 같은 namespace로 부를지 정해야 한다.
실행 기록과 시험은 합성 범위에서 읽는다
이번 패키지에는 27개 시험 메서드와 다섯 상황의 로컬 실행 기록을 넣었다. 같은 ID의 제목 변경, 같은 제목의 다른 ID, 식별자 누락, 두 정본 후보 사이의 미확인 행, 빈 제목 키를 다룬다. 시험은 세 행의 여섯 순열도 비교한다. 27이라는 수에 그 순열과 하위 입력 조합을 별도 시험으로 더하지 않았다.
아래 선택기는 records.json의 저장된 결과를 보여 준다. 방문자의 브라우저에서 Python을 실행하거나 글을 수집하는 화면은 아니다. ZIP 추출 뒤 같은 시험을 실행할 수 있고 make_records.py로 기록을 다시 만들 수 있다. 원래 수집기와 실제 검색 서비스의 동작 검증, 렌더링 화면·키보드 검수는 이 합성 시험의 범위에 들어가지 않는다.
다른 프로그램에 연결할 때 남은 책임
압축을 푼 폴더에서 python -B -m unittest -v를 실행하고, Python 코드에서는 identity_review의 project 함수를 불러 합성 행 목록을 전달하면 된다. 파일·서버·브라우저 연결은 없다. 실제 자료를 쓰려면 신뢰할 수 있는 어댑터에서 식별자를 확인하고 이 예제의 필드에 맞는 별칭을 만드는 과정부터 설계해야 한다. 제목 비교만으로 삭제·합치기 버튼을 실행하지 않는다.
출력의 identity_authenticated, normalized_title_proves_identity, merge_authorized, performs_external_write는 항상 false다. 출처별 검색 관측 상태가 필요하면 아래 기존 모듈도 별도로 살펴볼 수 있다. 두 모듈을 자동 연결한 것은 아니며, 어느 글인지 검토하는 책임과 그 글을 실제로 조사했는지 표현하는 책임을 나눠 재사용할 수 있게 남겼다.