-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcli.py
More file actions
executable file
·183 lines (163 loc) · 6.55 KB
/
Copy pathcli.py
File metadata and controls
executable file
·183 lines (163 loc) · 6.55 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
import click
from database import init_db, add_file, create_folder, rename_item, add_vip_user, remove_vip_user, get_all_vip_users, move_items, copy_items, restore_items, permanent_delete_items, empty_recycle_bin, delete_item # Moved delete_item to end
from user_handler import scan_channel_history
from config import TELEGRAM_CHAT_ID
import asyncio
import os
import json
import base64 # Added for decoding JSON from app.py
@click.group()
def cli():
pass
@click.command()
def initdb():
"""Initialize the database."""
init_db()
click.echo('Initialized the database.')
@click.command()
@click.option('--chat_id', default=TELEGRAM_CHAT_ID, help='Telegram Chat ID to scan.')
def scan_history(chat_id):
"""Scan Telegram channel/group history for files."""
click.echo(f'Starting to scan history for chat ID: {chat_id}')
asyncio.run(scan_channel_history(chat_id, add_file))
click.echo('History scan complete.')
@click.command()
@click.argument('folder_path')
def create_folder_cli(folder_path):
"""Create a new folder in the drive."""
from app import app # Import app here to avoid circular dependency
with app.app_context():
create_folder(folder_path)
click.echo(f'Folder "{folder_path}" created.')
@click.command()
@click.argument('item_id') # This will be the DB ID for files, or path for folders
@click.argument('new_name')
@click.option('--is_folder', is_flag=True, help='Set if renaming a folder.')
@click.option('--current_folder', default=None, help='Current folder path for folder renaming.')
def rename(item_id, new_name, is_folder, current_folder):
"""Rename a file or folder."""
from app import app
with app.app_context():
# item_id is already the correct DB ID for files, or path for folders
rename_item(item_id, new_name, is_folder=is_folder, current_folder=current_folder)
click.echo(f'Renamed "{item_id}" to "{new_name}".')
# Removed the old 'move' command as it's replaced by 'move_bulk' and the logic is now in database.py
@click.command()
@click.argument('item_id') # This will be the DB ID for files, or path for folders
@click.option('--is_folder', is_flag=True, help='Set if deleting a folder.') # Keep for now to differentiate single file/folder delete
def delete(item_id, is_folder): # Removed db_only as it's now always logical delete
"""Delete a file or folder (moves to recycle bin)."""
from app import app
with app.app_context():
# For single delete, item_id is the DB ID for files, or path for folders
# The delete_item function in database.py now handles the logic based on item_type
# We need to pass a list of dictionaries for delete_item to handle it uniformly
if is_folder:
# For folders, item_id is the folder path
items_to_delete = [{'id': item_id, 'type': 'folder'}]
else:
# For files, item_id is the DB ID
items_to_delete = [{'id': int(item_id), 'type': 'file'}] # Ensure ID is int for files
delete_item(items_to_delete, is_bulk=True) # Treat single delete as bulk for consistency
click.echo(f'Moved "{item_id}" to recycle bin.')
@click.command()
@click.argument('user_id', type=int)
def add_vip(user_id):
"""Add a user to the VIP list."""
from app import app
with app.app_context():
add_vip_user(user_id)
click.echo(f'User {user_id} added as VIP.')
@click.command()
@click.argument('user_id', type=int)
def remove_vip(user_id):
"""Remove a user from the VIP list."""
from app import app
with app.app_context():
remove_vip_user(user_id)
click.echo(f'User {user_id} removed from VIPs.')
@click.command()
def list_vips():
"""List all VIP users."""
from app import app
with app.app_context():
vips = get_all_vip_users()
if vips:
click.echo("VIP Users:")
for user_id in vips:
click.echo(f'- {user_id}')
else:
click.echo("No VIP users found.")
@click.command()
@click.argument('items_json')
@click.argument('destination_folder')
def move_bulk(items_json, destination_folder):
"""Move multiple files or folders."""
decoded_items_json = base64.b64decode(items_json).decode('utf-8')
items = json.loads(decoded_items_json)
from app import app
with app.app_context():
move_items(items, destination_folder)
click.echo(f'Moved {len(items)} items to "{destination_folder}".')
@click.command()
@click.argument('items_json')
@click.argument('destination_folder')
def copy_bulk(items_json, destination_folder):
"""Copy multiple files or folders."""
decoded_items_json = base64.b64decode(items_json).decode('utf-8')
items = json.loads(decoded_items_json)
from app import app
with app.app_context():
copy_items(items, destination_folder)
click.echo(f'Copied {len(items)} items to "{destination_folder}".')
@click.command()
@click.argument('items_json')
def delete_bulk(items_json): # Removed db_only as it's now always logical delete
"""Delete multiple files or folders (moves to recycle bin)."""
items = json.loads(items_json)
from app import app
with app.app_context():
delete_item(items, is_bulk=True)
click.echo(f'Moved {len(items)} items to recycle bin.')
@click.command()
@click.argument('item_ids_json')
def restore_bulk(item_ids_json):
"""Restore multiple items from the recycle bin."""
item_ids = json.loads(item_ids_json)
from app import app
with app.app_context():
restore_items(item_ids)
click.echo(f'Restored {len(item_ids)} items from recycle bin.')
@click.command()
@click.argument('item_ids_json')
def permanent_delete_bulk(item_ids_json):
"""Permanently delete multiple items from the recycle bin."""
item_ids = json.loads(item_ids_json)
from app import app
with app.app_context():
permanent_delete_items(item_ids)
click.echo(f'Permanently deleted {len(item_ids)} items.')
@click.command()
def empty_recycle_bin():
"""Empty the recycle bin, permanently deleting all items."""
from app import app
with app.app_context():
empty_recycle_bin()
click.echo('Recycle bin emptied.')
cli.add_command(initdb)
cli.add_command(scan_history)
cli.add_command(create_folder_cli)
cli.add_command(rename)
# cli.add_command(move) # Removed
cli.add_command(delete)
cli.add_command(add_vip)
cli.add_command(remove_vip)
cli.add_command(list_vips)
cli.add_command(move_bulk)
cli.add_command(copy_bulk)
cli.add_command(delete_bulk)
cli.add_command(restore_bulk) # Added
cli.add_command(permanent_delete_bulk) # Added
cli.add_command(empty_recycle_bin) # Added
if __name__ == '__main__':
cli()