| """Training-only mirror generation with a user-configured local model command. |
| Input JSONL: verified human documents with text/group/domain/provenance. |
| Command accepts one JSON prompt on stdin and returns {text: ...} on stdout. |
| No shell, no hidden remote requests, and no exposed paper in this pipeline. |
| """ |
| import argparse,hashlib,json,shlex,subprocess |
| from pathlib import Path |
|
|
| def main(): |
| p=argparse.ArgumentParser();p.add_argument('input',type=Path);p.add_argument('--output',required=True,type=Path);p.add_argument('--command',required=True);p.add_argument('--generator',required=True);p.add_argument('--temperature',type=float,default=.8);a=p.parse_args() |
| with a.output.open('w') as out: |
| for line in a.input.read_text().splitlines(): |
| r=json.loads(line) |
| if r.get('label')!='human' or not r.get('provenance'):raise ValueError('Only provenance-verified human training inputs accepted.') |
| if r.get('split')!='train':raise ValueError('Never generate training mirrors from calibration or final test documents.') |
| prompt={'task':'Generate a training counterpart matching topic, register and approximate length. Return only the new prose; do not reproduce the original verbatim.','source':r['text'],'target_words':len(r['text'].split()),'temperature':a.temperature,'generator':a.generator} |
| process=subprocess.run(shlex.split(a.command),input=json.dumps(prompt),text=True,capture_output=True,timeout=180,check=True) |
| response=json.loads(process.stdout);text=response['text'].strip() |
| if not text or text==r['text'].strip():raise ValueError('Invalid mirror: empty or verbatim copy.') |
| output=r|{'text':text,'label':'ai','generator':a.generator,'source_sha256':hashlib.sha256(r['text'].encode()).hexdigest(),'generation_settings':{'temperature':a.temperature},'mirror_word_ratio':len(text.split())/max(1,len(r['text'].split()))} |
| out.write(json.dumps(output)+'\n') |
| if __name__=='__main__':main() |
|
|
