73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
import click
|
|
from werkzeug.security import generate_password_hash
|
|
from app.extensions import db
|
|
from app.models.user import User
|
|
from app.models.company import Company, UserCompanyAccess
|
|
from app.seed import seed_data
|
|
from app.services.backup_service import BackupService
|
|
from app.services.mail_service import MailService
|
|
from app.services.pdf_service import PdfService
|
|
from app.services.sync_service import SyncService
|
|
|
|
|
|
def register_cli(app):
|
|
@app.cli.command('init-db')
|
|
def init_db():
|
|
db.create_all()
|
|
print('Database initialized')
|
|
|
|
@app.cli.command('create-company')
|
|
@click.option('--name', prompt=True)
|
|
@click.option('--tax-id', default='')
|
|
def create_company(name, tax_id):
|
|
company = Company(name=name, tax_id=tax_id)
|
|
db.session.add(company)
|
|
db.session.commit()
|
|
print(f'Company created: {company.id}')
|
|
|
|
@app.cli.command('create-user')
|
|
@click.option('--email', prompt=True)
|
|
@click.option('--name', prompt=True)
|
|
@click.option('--password', prompt=True, hide_input=True, confirmation_prompt=True)
|
|
@click.option('--role', default='admin', type=click.Choice(['admin', 'operator', 'readonly']))
|
|
@click.option('--company-id', type=int, default=None)
|
|
@click.option('--access-level', default='full', type=click.Choice(['full', 'readonly']))
|
|
def create_user(email, name, password, role, company_id, access_level):
|
|
if User.query.filter_by(email=email.lower()).first():
|
|
raise click.ClickException('User already exists')
|
|
user = User(email=email.lower(), name=name, password_hash=generate_password_hash(password), role=role)
|
|
db.session.add(user)
|
|
db.session.flush()
|
|
if company_id:
|
|
db.session.add(UserCompanyAccess(user_id=user.id, company_id=company_id, access_level=access_level))
|
|
db.session.commit()
|
|
print('User created')
|
|
|
|
@app.cli.command('seed-data')
|
|
def seed_command():
|
|
seed_data()
|
|
print('Seed data inserted')
|
|
|
|
@app.cli.command('sync-ksef')
|
|
@click.option('--company-id', type=int, default=None)
|
|
def sync_ksef(company_id):
|
|
company = db.session.get(Company, company_id) if company_id else Company.query.first()
|
|
SyncService(company).run_manual_sync()
|
|
print('KSeF sync complete')
|
|
|
|
@app.cli.command('rebuild-pdf')
|
|
def rebuild_pdf():
|
|
PdfService().rebuild_all()
|
|
print('PDF rebuild complete')
|
|
|
|
@app.cli.command('resend-mail')
|
|
@click.option('--delivery-id', prompt=True, type=int)
|
|
def resend_mail(delivery_id):
|
|
MailService().retry_delivery(delivery_id)
|
|
print('Mail resent')
|
|
|
|
@app.cli.command('backup-data')
|
|
def backup_data():
|
|
path = BackupService().create_backup()
|
|
print(f'Backup created: {path}')
|