import gradio as gr import json import requests import pandas as pd import traceback # --- LLM Configuration --- LLM_API_KEY = "sk-fa6c38ce957e4c7b946ccbeed33237ec" LLM_API_URL = "https://api.deepseek.com/v1/chat/completions" def call_llm(prompt, system_prompt="You are a helpful assistant."): headers = { "Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json" } data = { "model": "deepseek-chat", "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], "stream": False } try: response = requests.post(LLM_API_URL, headers=headers, json=data, timeout=60) response.raise_for_status() return response.json()['choices'][0]['message']['content'] except Exception as e: return f"Error: {str(e)}" def analyze_json_structure(json_input): try: # Try parsing as JSON data = json.loads(json_input) except: # Try parsing as JSONL (first line) try: data = json.loads(json_input.strip().split('\n')[0]) except Exception as e: return [], f"Parse Error: {e}" prompt = f""" Analyze this JSON item from an SFT dataset: {json.dumps(data, indent=2)} 1. Identify all fields, their types, and a short sample value. 2. For each field, suggest 1-3 common data cleaning/modification actions relevant to SFT (e.g., "Normalize score", "Remove 'User:' prefix", "Fix HTML entities", "Delete if empty"). 3. Return ONLY a JSON list of objects with keys: "field", "type", "sample", "suggestions" (list of strings). """ response = call_llm(prompt, "You are a data engineering expert.") # Clean response if "```json" in response: response = response.split("```json")[1].split("```")[0] elif "```" in response: response = response.split("```")[1].split("```")[0] try: analysis = json.loads(response.strip()) # Convert to list of dicts for DataFrame return analysis, "Analysis Complete" except Exception as e: return [], f"LLM Parse Error: {e}\nRaw: {response}" def generate_transform_code(json_sample, rules): # rules is a list of dicts: [{'field': 'x', 'action': 'y', 'custom': 'z'}] prompt = f""" I have a JSON item structure like this: {json_sample} I need a Python function `transform(item)` that modifies this item based on these rules: {json.dumps(rules, indent=2)} Requirements: 1. The function must take a dict `item` and return the modified dict. 2. If the item should be filtered out (dropped), return None. 3. Handle missing fields gracefully. 4. Return ONLY the Python code for the function. No markdown. """ code = call_llm(prompt, "You are a Python expert.") if "```python" in code: code = code.split("```python")[1].split("```")[0] elif "```" in code: code = code.split("```")[1].split("```")[0] return code.strip() def generate_full_script(transform_code): template = f"""import orjson import tqdm import argparse import sys def transform(item): {transform_code} def main(): parser = argparse.ArgumentParser() parser.add_argument('--input', required=True, help='Input JSON/JSONL file') parser.add_argument('--output', required=True, help='Output JSONL file') args = parser.parse_args() print(f"Processing {{args.input}} -> {{args.output}}") with open(args.input, 'rb') as f_in, open(args.output, 'wb') as f_out: # Detect format roughly first_char = f_in.read(1) f_in.seek(0) is_jsonl = True # Default assumption or logic # Simple line-by-line processing for JSONL # For standard JSON list, we'd need ijson or similar for streaming, # but for simplicity let's assume JSONL or small JSON. lines = f_in if first_char == b'[': print("Warning: Standard JSON list detected. Loading full file (memory intensive).") data = orjson.loads(f_in.read()) lines = data is_jsonl = False processed_count = 0 for line in tqdm.tqdm(lines): if is_jsonl: try: item = orjson.loads(line) except: continue else: item = line result = transform(item) if result is not None: f_out.write(orjson.dumps(result) + b'\\n') processed_count += 1 print(f"Done. Wrote {{processed_count}} items.") if __name__ == "__main__": main() """ return template # --- UI Logic --- def on_analyze(json_text): analysis, msg = analyze_json_structure(json_text) # Prepare choices for dropdowns fields = [item['field'] for item in analysis] if analysis else [] # Store analysis in State return analysis, gr.update(choices=fields), msg def on_field_select(field, analysis_data): # Find suggestions for this field suggestions = ["Keep Unchanged", "Delete Field", "Custom"] if analysis_data: for item in analysis_data: if item['field'] == field: suggestions += item.get('suggestions', []) break # Ensure Custom is always available if "Custom" not in suggestions: suggestions.append("Custom") return gr.update(choices=suggestions, value=suggestions[0]) def add_rule(field, action, custom, current_rules): if not current_rules: current_rules = [] rule_desc = action if action == "Custom": rule_desc = f"Custom: {custom}" new_rule = {"field": field, "action": action, "custom": custom, "display": f"{field} -> {rule_desc}"} current_rules.append(new_rule) # Return updated dataframe data display_data = [[r['field'], r['action'], r['custom']] for r in current_rules] return current_rules, display_data def run_preview(json_text, rules): if not rules: return "No rules defined." # 1. Generate Code transform_code = generate_transform_code(json_text, rules) # 2. Execute locally (Safe-ish for this context) local_scope = {} try: exec(transform_code, {}, local_scope) transform_func = local_scope.get('transform') if not transform_func: return "Error: Could not find 'transform' function in generated code." # Parse input try: item = json.loads(json_text) except: item = json.loads(json_text.strip().split('\n')[0]) # Run result = transform_func(item) return { "original": item, "modified": result, "code": transform_code } except Exception as e: return f"Execution Error: {e}\nCode:\n{transform_code}" def create_ui(): gr.Markdown(""" ## ⚡ Fastest JSON Editor (快速 JSON 编辑器) Intelligent analysis and modification of JSON/JSONL data using LLM. 利用 LLM 智能分析和修改 JSON/JSONL 数据，生成高性能处理脚本。 """) with gr.Row(): with gr.Column(scale=1): json_input = gr.Textbox(label="Sample JSON Item", lines=10, placeholder="Paste a single JSON object here...") analyze_btn = gr.Button("🔍 Analyze Structure") status_msg = gr.Markdown("") with gr.Column(scale=1): # Field Inspector analysis_state = gr.State([]) rules_state = gr.State([]) with gr.Group(): gr.Markdown("### 🛠️ Add Modification Rule") field_dropdown = gr.Dropdown(label="Select Field", choices=[]) action_dropdown = gr.Dropdown(label="Action", choices=["Keep Unchanged", "Delete Field", "Custom"], allow_custom_value=True) custom_input = gr.Textbox(label="Custom Instruction (if needed)", placeholder="e.g. Convert to YYYY-MM-DD") add_btn = gr.Button("Add Rule") rules_table = gr.Dataframe(headers=["Field", "Action", "Custom"], label="Active Rules", interactive=False) with gr.Row(): preview_btn = gr.Button("▶️ Preview & Generate Code", variant="primary") with gr.Row(): with gr.Column(): preview_json = gr.JSON(label="Preview Result (Diff)") with gr.Column(): code_output = gr.Code(label="Generated Transform Function", language="python") with gr.Row(): gen_script_btn = gr.Button("🚀 Generate Full Script") full_script_output = gr.Code(label="Full Production Script", language="python", visible=False) # Event Wiring analyze_btn.click(on_analyze, inputs=[json_input], outputs=[analysis_state, field_dropdown, status_msg]) field_dropdown.change(on_field_select, inputs=[field_dropdown, analysis_state], outputs=[action_dropdown]) add_btn.click(add_rule, inputs=[field_dropdown, action_dropdown, custom_input, rules_state], outputs=[rules_state, rules_table]) preview_btn.click(run_preview, inputs=[json_input, rules_state], outputs=[preview_json]) # Update code output from preview result def update_code_view(result): if isinstance(result, dict): return result.get('code', '') return "" preview_btn.click(update_code_view, inputs=[preview_json], outputs=[code_output]) def on_gen_script(code): return gr.update(visible=True, value=generate_full_script(code)) gen_script_btn.click(on_gen_script, inputs=[code_output], outputs=[full_script_output])