cb463c0faa
Add per-site CSV and PDF matrix reports, editable sick-note date ranges, and weekend calendar styling. Preserve verified duplicate uploads for audit while excluding their entries from hours and costs, and prevent repeat files and overlapping daily imports.
138 lines
7.1 KiB
Python
138 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Audit exact duplicate uploads; --apply preserves files/hours and excludes duplicate entries."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import urllib.parse
|
|
import urllib.request
|
|
from collections import defaultdict
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pb_env import PB, http, superuser_token
|
|
|
|
|
|
def main():
|
|
apply = argparse.ArgumentParser()
|
|
apply.add_argument('--apply', action='store_true')
|
|
args = apply.parse_args()
|
|
token = superuser_token()
|
|
|
|
def request(method, path, body=None):
|
|
code, data = http(method, path, body, token)
|
|
if code not in (200, 204):
|
|
raise RuntimeError(f'{method} {path}: HTTP {code}')
|
|
return data
|
|
|
|
def records(collection):
|
|
rows, page = [], 1
|
|
while True:
|
|
result = request('GET', f'/api/collections/{collection}/records?perPage=500&page={page}&sort=id')
|
|
rows.extend(result['items'])
|
|
if page >= result['totalPages']:
|
|
return rows
|
|
page += 1
|
|
|
|
docs, entries = records('timesheets'), records('timesheet_entries')
|
|
schemas = {name: request('GET', '/api/collections/' + name) for name in ('timesheets', 'timesheet_entries')}
|
|
fd, backup = tempfile.mkstemp(prefix='mega-timesheet-dedup-', suffix='.json')
|
|
with os.fdopen(fd, 'w') as file:
|
|
json.dump({'schemas': schemas, 'documents': docs, 'entries': entries}, file)
|
|
print('Private backup:', backup)
|
|
file_token = request('POST', '/api/files/token', {})['token']
|
|
groups, hashes = defaultdict(list), {}
|
|
for doc in sorted(docs, key=lambda d: (d.get('date_uploaded', ''), d['id'])):
|
|
if not doc.get('source_file'):
|
|
continue
|
|
path = '/api/files/' + doc['collectionId'] + '/' + doc['id'] + '/' + urllib.parse.quote(doc['source_file'])
|
|
url = PB + path + '?' + urllib.parse.urlencode({'token': file_token})
|
|
req = urllib.request.Request(url, headers={'Authorization': token})
|
|
with urllib.request.urlopen(req) as response:
|
|
digest = hashlib.sha256(response.read()).hexdigest()
|
|
hashes[doc['id']] = digest
|
|
groups[(doc.get('site', ''), digest)].append(doc)
|
|
by_doc = defaultdict(list)
|
|
for entry in entries:
|
|
by_doc[entry.get('timesheet', '')].append(entry)
|
|
exclusions, duplicate_docs = {}, {}
|
|
for group in groups.values():
|
|
# Prefer the complete upload, then the earliest one. Empty/failed
|
|
# predecessors must not cause valid later entries to disappear.
|
|
canonical = max(group, key=lambda d: len(by_doc[d['id']]))
|
|
originals = defaultdict(list)
|
|
for row in by_doc[canonical['id']]:
|
|
originals[(row.get('employee'), row.get('site'), row.get('date_logged'), row.get('hours_worked'))].append(row['id'])
|
|
for doc in group:
|
|
if doc['id'] == canonical['id']:
|
|
continue
|
|
matched = defaultdict(int)
|
|
for row in by_doc[doc['id']]:
|
|
key = (row.get('employee'), row.get('site'), row.get('date_logged'), row.get('hours_worked'))
|
|
pos = matched[key]
|
|
if pos >= len(originals[key]):
|
|
raise RuntimeError('Same source file has different imported entries; review required before applying')
|
|
exclusions[row['id']] = originals[key][pos]
|
|
matched[key] += 1
|
|
duplicate_docs[doc['id']] = canonical['id']
|
|
print(json.dumps({'uploads': len(docs), 'entries': len(entries), 'duplicate_uploads': len(duplicate_docs), 'duplicate_entries': len(exclusions), 'excluded_hours': round(sum(r.get('hours_worked', 0) for r in entries if r['id'] in exclusions), 2)}))
|
|
if not args.apply:
|
|
return
|
|
keys = [':'.join((r['employee'], r['site'], r['date_logged'][:10])) for r in entries if r.get('timesheet') and r['id'] not in exclusions]
|
|
assert len(keys) == len(set(keys)), 'Distinct imported days overlap; review before applying'
|
|
additions = {
|
|
'timesheets': [{'name': 'import_key', 'type': 'text', 'required': False}, {'name': 'duplicate_of', 'type': 'text', 'required': False}],
|
|
'timesheet_entries': [{'name': 'excluded_duplicate', 'type': 'bool', 'required': False}, {'name': 'duplicate_of', 'type': 'text', 'required': False}, {'name': 'import_key', 'type': 'text', 'required': False}],
|
|
}
|
|
for name, schema in schemas.items():
|
|
fields = schema['fields'] + [f for f in additions[name] if not any(old['name'] == f['name'] for old in schema['fields'])]
|
|
request('PATCH', '/api/collections/' + schema['id'], {'fields': fields})
|
|
for doc in docs:
|
|
if doc['id'] not in hashes:
|
|
continue
|
|
duplicate = duplicate_docs.get(doc['id'], '')
|
|
request('PATCH', '/api/collections/timesheets/records/' + doc['id'], {'file_hash': hashes[doc['id']], 'duplicate_of': duplicate, 'import_key': '' if duplicate or not by_doc[doc['id']] else doc.get('site', '') + ':' + hashes[doc['id']]})
|
|
day_keys = set()
|
|
updates = []
|
|
for row in entries:
|
|
if row['id'] in exclusions:
|
|
changes = {'excluded_duplicate': True, 'duplicate_of': exclusions[row['id']], 'import_key': ''}
|
|
elif row.get('timesheet'):
|
|
key = ':'.join((row['employee'], row['site'], row['date_logged'][:10]))
|
|
assert key not in day_keys, 'Overlapping distinct imported days need review'
|
|
day_keys.add(key)
|
|
changes = {'import_key': key}
|
|
else:
|
|
continue
|
|
updates.append((row['id'], changes))
|
|
def update_entry(item):
|
|
ident, changes = item
|
|
request('PATCH', '/api/collections/timesheet_entries/records/' + ident, changes)
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
list(pool.map(update_entry, updates))
|
|
for name in schemas:
|
|
schema = request('GET', '/api/collections/' + name)
|
|
index = f'CREATE UNIQUE INDEX `idx_{name}_import_key` ON `{name}` (`import_key`) WHERE `import_key` != \'\''
|
|
if index not in schema['indexes']:
|
|
request('PATCH', '/api/collections/' + schema['id'], {'indexes': schema['indexes'] + [index]})
|
|
after_docs = records('timesheets')
|
|
assert len(after_docs) == len(docs)
|
|
old_docs = {d['id']: d for d in docs}
|
|
for doc in after_docs:
|
|
for key, value in old_docs[doc['id']].items():
|
|
if key not in ('updated', 'file_hash', 'duplicate_of', 'import_key'):
|
|
assert doc.get(key) == value, f'Original document changed: {key}'
|
|
after = records('timesheet_entries')
|
|
assert len(after) == len(entries)
|
|
originals = {r['id']: r for r in entries}
|
|
for row in after:
|
|
before = originals[row['id']]
|
|
for key, value in before.items():
|
|
if key not in ('updated', 'excluded_duplicate', 'duplicate_of', 'import_key'):
|
|
assert row.get(key) == value, f'Original entry changed: {key}'
|
|
if row['id'] in exclusions:
|
|
assert row['excluded_duplicate'] and row['duplicate_of'] == exclusions[row['id']]
|
|
print('Verified: all original entry fields, hours and files preserved; duplicates flagged.')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|