335 lines
10 KiB
Python
Executable File
335 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
LabArchives Entry Operations
|
|
|
|
Utilities for creating entries, uploading attachments, and managing notebook content.
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
import yaml
|
|
import os
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
|
|
def load_config(config_path='config.yaml'):
|
|
"""Load configuration from YAML file"""
|
|
try:
|
|
with open(config_path, 'r') as f:
|
|
return yaml.safe_load(f)
|
|
except FileNotFoundError:
|
|
print(f"❌ Configuration file not found: {config_path}")
|
|
print(" Run setup_config.py first to create configuration")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"❌ Error loading configuration: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
def init_client(config):
|
|
"""Initialize LabArchives API client"""
|
|
try:
|
|
from labarchivespy.client import Client
|
|
return Client(
|
|
config['api_url'],
|
|
config['access_key_id'],
|
|
config['access_password']
|
|
)
|
|
except ImportError:
|
|
print("❌ labarchives-py package not installed")
|
|
print(" Install with: pip install git+https://github.com/mcmero/labarchives-py")
|
|
sys.exit(1)
|
|
|
|
|
|
def get_user_id(client, config):
|
|
"""Get user ID via authentication"""
|
|
import xml.etree.ElementTree as ET
|
|
|
|
login_params = {
|
|
'login_or_email': config['user_email'],
|
|
'password': config['user_external_password']
|
|
}
|
|
|
|
try:
|
|
response = client.make_call('users', 'user_access_info', params=login_params)
|
|
|
|
if response.status_code == 200:
|
|
uid = ET.fromstring(response.content)[0].text
|
|
return uid
|
|
else:
|
|
print(f"❌ Authentication failed: HTTP {response.status_code}")
|
|
print(f" Response: {response.content.decode('utf-8')[:200]}")
|
|
sys.exit(1)
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error during authentication: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
def create_entry(client, uid, nbid, title, content=None, date=None):
|
|
"""Create a new entry in a notebook"""
|
|
print(f"\n📝 Creating entry: {title}")
|
|
|
|
# Prepare parameters
|
|
params = {
|
|
'uid': uid,
|
|
'nbid': nbid,
|
|
'title': title
|
|
}
|
|
|
|
if content:
|
|
# Ensure content is HTML formatted
|
|
if not content.startswith('<'):
|
|
content = f'<p>{content}</p>'
|
|
params['content'] = content
|
|
|
|
if date:
|
|
params['date'] = date
|
|
|
|
try:
|
|
response = client.make_call('entries', 'create_entry', params=params)
|
|
|
|
if response.status_code == 200:
|
|
print("✅ Entry created successfully")
|
|
|
|
# Try to extract entry ID from response
|
|
try:
|
|
import xml.etree.ElementTree as ET
|
|
root = ET.fromstring(response.content)
|
|
entry_id = root.find('.//entry_id')
|
|
if entry_id is not None:
|
|
print(f" Entry ID: {entry_id.text}")
|
|
return entry_id.text
|
|
except:
|
|
pass
|
|
|
|
return True
|
|
|
|
else:
|
|
print(f"❌ Entry creation failed: HTTP {response.status_code}")
|
|
print(f" Response: {response.content.decode('utf-8')[:200]}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error creating entry: {e}")
|
|
return None
|
|
|
|
|
|
def create_comment(client, uid, nbid, entry_id, comment):
|
|
"""Add a comment to an existing entry"""
|
|
print(f"\n💬 Adding comment to entry {entry_id}")
|
|
|
|
params = {
|
|
'uid': uid,
|
|
'nbid': nbid,
|
|
'entry_id': entry_id,
|
|
'comment': comment
|
|
}
|
|
|
|
try:
|
|
response = client.make_call('entries', 'create_comment', params=params)
|
|
|
|
if response.status_code == 200:
|
|
print("✅ Comment added successfully")
|
|
return True
|
|
else:
|
|
print(f"❌ Comment creation failed: HTTP {response.status_code}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error creating comment: {e}")
|
|
return False
|
|
|
|
|
|
def upload_attachment(client, config, uid, nbid, entry_id, file_path):
|
|
"""Upload a file attachment to an entry"""
|
|
import requests
|
|
|
|
file_path = Path(file_path)
|
|
|
|
if not file_path.exists():
|
|
print(f"❌ File not found: {file_path}")
|
|
return False
|
|
|
|
print(f"\n📎 Uploading attachment: {file_path.name}")
|
|
print(f" Size: {file_path.stat().st_size / 1024:.2f} KB")
|
|
|
|
url = f"{config['api_url']}/entries/upload_attachment"
|
|
|
|
try:
|
|
with open(file_path, 'rb') as f:
|
|
files = {'file': f}
|
|
data = {
|
|
'uid': uid,
|
|
'nbid': nbid,
|
|
'entry_id': entry_id,
|
|
'filename': file_path.name,
|
|
'access_key_id': config['access_key_id'],
|
|
'access_password': config['access_password']
|
|
}
|
|
|
|
response = requests.post(url, files=files, data=data)
|
|
|
|
if response.status_code == 200:
|
|
print("✅ Attachment uploaded successfully")
|
|
return True
|
|
else:
|
|
print(f"❌ Upload failed: HTTP {response.status_code}")
|
|
print(f" Response: {response.content.decode('utf-8')[:200]}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error uploading attachment: {e}")
|
|
return False
|
|
|
|
|
|
def batch_upload(client, config, uid, nbid, entry_id, directory):
|
|
"""Upload all files from a directory as attachments"""
|
|
directory = Path(directory)
|
|
|
|
if not directory.is_dir():
|
|
print(f"❌ Directory not found: {directory}")
|
|
return
|
|
|
|
files = list(directory.glob('*'))
|
|
files = [f for f in files if f.is_file()]
|
|
|
|
if not files:
|
|
print(f"❌ No files found in {directory}")
|
|
return
|
|
|
|
print(f"\n📦 Batch uploading {len(files)} files from {directory}")
|
|
|
|
successful = 0
|
|
failed = 0
|
|
|
|
for file_path in files:
|
|
if upload_attachment(client, config, uid, nbid, entry_id, file_path):
|
|
successful += 1
|
|
else:
|
|
failed += 1
|
|
|
|
print("\n" + "="*60)
|
|
print(f"Batch upload complete: {successful} successful, {failed} failed")
|
|
print("="*60)
|
|
|
|
|
|
def create_entry_with_attachments(client, config, uid, nbid, title, content,
|
|
attachments):
|
|
"""Create entry and upload multiple attachments"""
|
|
# Create entry
|
|
entry_id = create_entry(client, uid, nbid, title, content)
|
|
|
|
if not entry_id:
|
|
print("❌ Cannot upload attachments without entry ID")
|
|
return False
|
|
|
|
# Upload attachments
|
|
for attachment_path in attachments:
|
|
upload_attachment(client, config, uid, nbid, entry_id, attachment_path)
|
|
|
|
return True
|
|
|
|
|
|
def main():
|
|
"""Main command-line interface"""
|
|
parser = argparse.ArgumentParser(
|
|
description='LabArchives Entry Operations',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Create simple entry
|
|
python3 entry_operations.py create --nbid 12345 --title "Experiment Results"
|
|
|
|
# Create entry with content
|
|
python3 entry_operations.py create --nbid 12345 --title "Results" \\
|
|
--content "PCR amplification successful"
|
|
|
|
# Create entry with HTML content
|
|
python3 entry_operations.py create --nbid 12345 --title "Results" \\
|
|
--content "<p>Results:</p><ul><li>Sample A: Positive</li></ul>"
|
|
|
|
# Upload attachment to existing entry
|
|
python3 entry_operations.py upload --nbid 12345 --entry-id 67890 \\
|
|
--file data.csv
|
|
|
|
# Batch upload multiple files
|
|
python3 entry_operations.py batch-upload --nbid 12345 --entry-id 67890 \\
|
|
--directory ./experiment_data/
|
|
|
|
# Add comment to entry
|
|
python3 entry_operations.py comment --nbid 12345 --entry-id 67890 \\
|
|
--text "Follow-up analysis needed"
|
|
"""
|
|
)
|
|
|
|
parser.add_argument('--config', default='config.yaml',
|
|
help='Path to configuration file (default: config.yaml)')
|
|
parser.add_argument('--nbid', required=True,
|
|
help='Notebook ID')
|
|
|
|
subparsers = parser.add_subparsers(dest='command', help='Command to execute')
|
|
|
|
# Create entry command
|
|
create_parser = subparsers.add_parser('create', help='Create new entry')
|
|
create_parser.add_argument('--title', required=True, help='Entry title')
|
|
create_parser.add_argument('--content', help='Entry content (HTML supported)')
|
|
create_parser.add_argument('--date', help='Entry date (YYYY-MM-DD)')
|
|
create_parser.add_argument('--attachments', nargs='+',
|
|
help='Files to attach to the new entry')
|
|
|
|
# Upload attachment command
|
|
upload_parser = subparsers.add_parser('upload', help='Upload attachment to entry')
|
|
upload_parser.add_argument('--entry-id', required=True, help='Entry ID')
|
|
upload_parser.add_argument('--file', required=True, help='File to upload')
|
|
|
|
# Batch upload command
|
|
batch_parser = subparsers.add_parser('batch-upload',
|
|
help='Upload all files from directory')
|
|
batch_parser.add_argument('--entry-id', required=True, help='Entry ID')
|
|
batch_parser.add_argument('--directory', required=True,
|
|
help='Directory containing files to upload')
|
|
|
|
# Comment command
|
|
comment_parser = subparsers.add_parser('comment', help='Add comment to entry')
|
|
comment_parser.add_argument('--entry-id', required=True, help='Entry ID')
|
|
comment_parser.add_argument('--text', required=True, help='Comment text')
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
# Load configuration and initialize
|
|
config = load_config(args.config)
|
|
client = init_client(config)
|
|
uid = get_user_id(client, config)
|
|
|
|
# Execute command
|
|
if args.command == 'create':
|
|
if args.attachments:
|
|
create_entry_with_attachments(
|
|
client, config, uid, args.nbid, args.title,
|
|
args.content, args.attachments
|
|
)
|
|
else:
|
|
create_entry(client, uid, args.nbid, args.title,
|
|
args.content, args.date)
|
|
|
|
elif args.command == 'upload':
|
|
upload_attachment(client, config, uid, args.nbid,
|
|
args.entry_id, args.file)
|
|
|
|
elif args.command == 'batch-upload':
|
|
batch_upload(client, config, uid, args.nbid,
|
|
args.entry_id, args.directory)
|
|
|
|
elif args.command == 'comment':
|
|
create_comment(client, uid, args.nbid, args.entry_id, args.text)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|