import base64 import datetime import hashlib import hmac import json import mimetypes import tempfile import time import uuid from pathlib import Path from urllib.parse import urlencode from odoo import http, _ from odoo.exceptions import AccessError, UserError from odoo.http import request TEMP_SUBDIR = 'odoo_ifc_temp' TEMP_TTL_MINUTES = 60 def _json_http_response(payload, status=200): return request.make_response( json.dumps(payload), headers=[('Content-Type', 'application/json')], status=status, ) def _jsonrpc_ok(result=None): return {'success': True, 'result': result or {}} def _jsonrpc_fail(message, code='ERROR', details=None): payload = {'success': False, 'error': {'code': code, 'message': message}} if details: payload['error']['details'] = details return payload def _safe_filename(filename, fallback='document.ifc'): name = (filename or fallback).strip().replace('\\', '_').replace('/', '_') return name or fallback def _to_bool(value, default=True): if isinstance(value, bool): return value if value is None: return default return str(value).strip().lower() in ('1', 'true', 'yes', 'y', 'on') class IfcJsonRpcController(http.Controller): def _check_record_access(self, record, mode='read'): record.check_access_rights(mode) record.check_access_rule(mode) def _serialize_project(self, project): return { 'id': project.id, 'name': project.name, 'document_ref': project.document_ref, 'file_name': project.file_name, 'company_id': project.company_id.id if project.company_id else False, 'company_name': project.company_id.name if project.company_id else False, 'state': project.state, 'approval_state': project.approval_state, 'uploaded_date': project.uploaded_date.isoformat() if project.uploaded_date else False, 'uploaded_by': { 'id': project.uploaded_by_id.id, 'name': project.uploaded_by_id.name, } if project.uploaded_by_id else False, 'last_update': project.last_update.isoformat() if project.last_update else False, 'last_revision_label': project.last_revision_label, 'last_revision_note': project.last_revision_note, 'last_revision_date': project.last_revision_date.isoformat() if project.last_revision_date else False, 'revised_by': { 'id': project.revised_by_id.id, 'name': project.revised_by_id.name, } if project.revised_by_id else False, 'estimated_bom_id': project.estimated_bom_id.id if project.estimated_bom_id else False, 'permanent_bom_id': project.permanent_bom_id.id if project.permanent_bom_id else False, 'latest_mo_id': project.manufacturing_order_id.id if project.manufacturing_order_id else False, 'mo_count': len(project.manufacturing_order_ids), 'activity_count': len(project.activity_ids), 'history_count': len(project.history_ids), 'revision_count': len(project.revision_ids), } def _serialize_activity(self, activity): return { 'id': activity.id, 'summary': activity.summary, 'note': activity.note, 'date_deadline': activity.date_deadline.isoformat() if activity.date_deadline else False, 'state': activity.state, 'res_id': activity.res_id, 'assigned_to': { 'id': activity.user_id.id, 'name': activity.user_id.name, } if activity.user_id else False, 'created_at': activity.create_date.isoformat() if activity.create_date else False, } def _serialize_history(self, history): return { 'id': history.id, 'activity_type': history.activity_type, 'activity_date': history.activity_date.isoformat() if history.activity_date else False, 'note': history.note, 'state_snapshot': history.state_snapshot, 'user': { 'id': history.user_id.id, 'name': history.user_id.name, } if history.user_id else False, } def _serialize_bom_line(self, line): return { 'id': line.id, 'name': line.product_id.display_name, 'product_id': line.product_id.id, 'default_code': line.product_id.default_code, 'qty': line.product_qty, 'uom': { 'id': line.product_uom_id.id, 'name': line.product_uom_id.name, } if line.product_uom_id else False, } def _serialize_bom(self, bom): return { 'id': bom.id, 'code': bom.code, 'product': { 'id': bom.product_tmpl_id.id, 'name': bom.product_tmpl_id.display_name, 'default_code': bom.product_tmpl_id.default_code, } if bom.product_tmpl_id else False, 'quantity': bom.product_qty, 'type': bom.type, 'line_count': len(bom.line_ids), 'lines': [self._serialize_bom_line(line) for line in bom.line_ids], } def _serialize_purchase_order(self, purchase_order): return { 'id': purchase_order.id, 'name': purchase_order.name, 'state': purchase_order.state, 'vendor': { 'id': purchase_order.partner_id.id, 'name': purchase_order.partner_id.name, } if purchase_order.partner_id else False, 'origin': purchase_order.origin, 'line_count': len(purchase_order.order_line), 'lines': [ { 'id': line.id, 'product_id': line.product_id.id, 'product_name': line.product_id.display_name, 'qty': line.product_qty, 'uom_id': line.product_uom.id, 'uom_name': line.product_uom.name, 'price_unit': line.price_unit, } for line in purchase_order.order_line ], } def _serialize_manufacturing_order(self, manufacturing_order): return { 'id': manufacturing_order.id, 'name': manufacturing_order.name, 'state': manufacturing_order.state, 'origin': manufacturing_order.origin, 'product': { 'id': manufacturing_order.product_id.id, 'name': manufacturing_order.product_id.display_name, 'default_code': manufacturing_order.product_id.default_code, } if manufacturing_order.product_id else False, 'qty': manufacturing_order.product_qty, 'uom': { 'id': manufacturing_order.product_uom_id.id, 'name': manufacturing_order.product_uom_id.name, } if manufacturing_order.product_uom_id else False, 'bom_id': manufacturing_order.bom_id.id if manufacturing_order.bom_id else False, 'ifc_project_id': manufacturing_order.ifc_project_id.id if manufacturing_order.ifc_project_id else False, } def _get_signing_secret(self): param = request.env['ir.config_parameter'].sudo() secret = param.get_param('database.secret') or param.get_param('database.uuid') return secret or request.session.db or 'odoo' def _build_temp_signature(self, file_key, document_id, user_id, expires_at): payload = f'{file_key}|{document_id}|{user_id}|{expires_at}' return hmac.new( self._get_signing_secret().encode('utf-8'), payload.encode('utf-8'), hashlib.sha256, ).hexdigest() def _build_signed_temp_download_url(self, file_key, document_id, user_id): expires_at = int(time.time()) + (TEMP_TTL_MINUTES * 60) signature = self._build_temp_signature(file_key, document_id, user_id, expires_at) query = urlencode( { 'doc_id': int(document_id), 'uid': int(user_id), 'exp': expires_at, 'sig': signature, } ) return f'/api/ifc/temp/{file_key}?{query}', expires_at def _verify_temp_signature(self, file_key, document_id, user_id, expires_at, signature): if not signature: return False try: exp_int = int(expires_at) except (TypeError, ValueError): return False if exp_int < int(time.time()): return False expected = self._build_temp_signature(file_key, document_id, user_id, exp_int) return hmac.compare_digest(expected, signature) def _get_project_for_action(self, document_id, mode='write'): project = request.env['grt.ifc.project'].browse(int(document_id)).exists() if not project: return False, _jsonrpc_fail('Dokumen tidak ditemukan.', code='NOT_FOUND') try: self._check_record_access(project, mode=mode) except AccessError: return False, _jsonrpc_fail('Tidak punya akses ke dokumen.', code='ACCESS_DENIED') return project, None def _execute_project_action(self, document_id, method_name): project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: getattr(project, method_name)() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='ACTION_FAILED') except Exception as exc: return _jsonrpc_fail('Eksekusi aksi dokumen gagal.', code='ACTION_FAILED', details=str(exc)) return _jsonrpc_ok({'document': self._serialize_project(project)}) def _get_temp_root(self): root = Path(tempfile.gettempdir()) / TEMP_SUBDIR root.mkdir(parents=True, exist_ok=True) return root def _cleanup_temp_files(self): root = self._get_temp_root() now = datetime.datetime.now(datetime.UTC) for file_path in root.glob('*.ifc'): try: age_seconds = now.timestamp() - file_path.stat().st_mtime except OSError: continue if age_seconds > TEMP_TTL_MINUTES * 60: try: file_path.unlink() except OSError: pass @http.route('/api/session/login', type='json', auth='none', methods=['POST'], csrf=False) def api_session_login(self, db=None, login=None, password=None, **kwargs): database = db or request.session.db username = login passwd = password if not database or not username or not passwd: return _jsonrpc_fail('db, login, dan password wajib diisi.', code='BAD_REQUEST') try: uid = request.session.authenticate(database, username, passwd) except Exception as exc: return _jsonrpc_fail('Login gagal.', code='AUTH_FAILED', details=str(exc)) if not uid: return _jsonrpc_fail('Login gagal. Cek kredensial.', code='AUTH_FAILED') user = request.env['res.users'].sudo().browse(uid) return _jsonrpc_ok( { 'uid': uid, 'name': user.name, 'login': user.login, 'db': database, 'session_id': request.session.sid, } ) @http.route('/api/session/logout', type='json', auth='user', methods=['POST'], csrf=False) def api_session_logout(self, **kwargs): request.session.logout() return _jsonrpc_ok({'message': 'Logout berhasil.'}) @http.route('/api/session/me', type='json', auth='user', methods=['POST'], csrf=False) def api_session_me(self, **kwargs): user = request.env.user return _jsonrpc_ok( { 'uid': user.id, 'name': user.name, 'login': user.login, 'session_id': request.session.sid, 'db': request.session.db, } ) @http.route('/api/ifc/documents/list', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_list(self, limit=50, offset=0, search=None, **kwargs): domain = [] if search: domain = ['|', ('name', 'ilike', search), ('document_ref', 'ilike', search)] projects = request.env['grt.ifc.project'].search(domain, limit=int(limit), offset=int(offset), order='id desc') for project in projects: self._check_record_access(project, mode='read') return _jsonrpc_ok( { 'count': len(projects), 'items': [self._serialize_project(project) for project in projects], } ) @http.route('/api/ifc/documents/create', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_create(self, name=None, file_name=None, file_data=None, revision_note=None, company_id=None, **kwargs): if not name or not file_data: return _jsonrpc_fail('name dan file_data (base64) wajib diisi.', code='BAD_REQUEST') vals = { 'name': name, 'file_name': _safe_filename(file_name or '%s.ifc' % name), 'file_data': file_data, 'revision_note_input': revision_note or _('Dokumen awal diunggah via API.'), } if company_id: vals['company_id'] = int(company_id) try: project = request.env['grt.ifc.project'].create(vals) except Exception as exc: return _jsonrpc_fail('Gagal membuat dokumen IFC.', code='CREATE_FAILED', details=str(exc)) self._check_record_access(project, mode='read') return _jsonrpc_ok({'document': self._serialize_project(project)}) @http.route('/api/ifc/documents/revise', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_revise(self, document_id=None, file_data=None, file_name=None, revision_note=None, **kwargs): if not document_id or not file_data: return _jsonrpc_fail('document_id dan file_data (base64) wajib diisi.', code='BAD_REQUEST') project = request.env['grt.ifc.project'].browse(int(document_id)).exists() if not project: return _jsonrpc_fail('Dokumen tidak ditemukan.', code='NOT_FOUND') try: self._check_record_access(project, mode='write') except AccessError: return _jsonrpc_fail('Tidak punya akses untuk revisi dokumen.', code='ACCESS_DENIED') vals = { 'file_data': file_data, 'revision_note_input': revision_note or _('Revisi dokumen via API.'), } if file_name: vals['file_name'] = _safe_filename(file_name) try: project.write(vals) except Exception as exc: return _jsonrpc_fail('Gagal melakukan revisi dokumen.', code='REVISION_FAILED', details=str(exc)) return _jsonrpc_ok({'document': self._serialize_project(project)}) @http.route('/api/ifc/documents/parse', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_parse(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project = request.env['grt.ifc.project'].browse(int(document_id)).exists() if not project: return _jsonrpc_fail('Dokumen tidak ditemukan.', code='NOT_FOUND') try: self._check_record_access(project, mode='write') project.action_parse_ifc() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='PARSE_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal parse IFC.', code='PARSE_FAILED', details=str(exc)) return _jsonrpc_ok({'document': self._serialize_project(project)}) @http.route('/api/ifc/documents/approval/request-validation', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_request_validation(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') return self._execute_project_action(document_id, 'action_request_validation') @http.route('/api/ifc/documents/approval/submit', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_submit_approval(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') return self._execute_project_action(document_id, 'action_submit_for_approval') @http.route('/api/ifc/documents/approval/approve-production', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_approve_production(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') return self._execute_project_action(document_id, 'action_approve_production') @http.route('/api/ifc/documents/approval/approve-provisioning', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_approve_provisioning(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') return self._execute_project_action(document_id, 'action_approve_provisioning') @http.route('/api/ifc/documents/approval/reject', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_reject(self, document_id=None, note=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: if note: project.revision_note_input = note project.action_reject_document() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='ACTION_FAILED') except Exception as exc: return _jsonrpc_fail('Reject dokumen gagal.', code='ACTION_FAILED', details=str(exc)) return _jsonrpc_ok({'document': self._serialize_project(project)}) @http.route('/api/ifc/documents/approval/back-draft', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_back_draft(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') return self._execute_project_action(document_id, 'action_back_to_draft') @http.route('/api/ifc/documents/activities/add', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_activities_add(self, document_id=None, summary=None, note=None, user_id=None, deadline=None, **kwargs): if not document_id or not summary: return _jsonrpc_fail('document_id dan summary wajib diisi.', code='BAD_REQUEST') project = request.env['grt.ifc.project'].browse(int(document_id)).exists() if not project: return _jsonrpc_fail('Dokumen tidak ditemukan.', code='NOT_FOUND') try: self._check_record_access(project, mode='write') except AccessError: return _jsonrpc_fail('Tidak punya akses membuat activity.', code='ACCESS_DENIED') target_user_id = int(user_id) if user_id else request.env.user.id date_deadline = deadline or False try: project.activity_schedule( 'mail.mail_activity_data_todo', summary=summary, note=note or '', user_id=target_user_id, date_deadline=date_deadline, ) project._log_history('validation', _('Activity baru ditambahkan via API.')) except Exception as exc: return _jsonrpc_fail('Gagal membuat activity.', code='ACTIVITY_FAILED', details=str(exc)) return _jsonrpc_ok({'message': 'Activity berhasil dibuat.'}) @http.route('/api/ifc/documents/activities/list', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_activities_list(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project = request.env['grt.ifc.project'].browse(int(document_id)).exists() if not project: return _jsonrpc_fail('Dokumen tidak ditemukan.', code='NOT_FOUND') try: self._check_record_access(project, mode='read') except AccessError: return _jsonrpc_fail('Tidak punya akses membaca activity.', code='ACCESS_DENIED') activities = request.env['mail.activity'].search( [ ('res_model', '=', 'grt.ifc.project'), ('res_id', '=', project.id), ], order='date_deadline asc, id desc', ) history = request.env['grt.ifc.activity.history'].search( [('project_id', '=', project.id)], order='activity_date desc, id desc', ) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'activities': [self._serialize_activity(activity) for activity in activities], 'history': [self._serialize_history(item) for item in history], } ) @http.route('/api/ifc/documents/bom/generate', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_bom_generate(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: project.action_create_bom_estimate() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='BOM_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal generate BoM estimasi.', code='BOM_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'estimated_bom': self._serialize_bom(project.estimated_bom_id) if project.estimated_bom_id else False, } ) @http.route('/api/ifc/documents/bom/get', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_bom_get(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='read') if error: return error parsed_bom = [ { 'id': line.id, 'ifc_key': line.ifc_key, 'name': line.name, 'global_id': line.global_id, 'ifc_type': line.ifc_type, 'relation_type': line.relation_type, 'level': line.level, 'quantity': line.quantity, 'unit_name': line.unit_name, 'parent_id': line.parent_id.id if line.parent_id else False, } for line in project.bom_ids ] return _jsonrpc_ok( { 'document': self._serialize_project(project), 'parsed_bom': parsed_bom, 'estimated_bom': self._serialize_bom(project.estimated_bom_id) if project.estimated_bom_id else False, 'permanent_bom': self._serialize_bom(project.permanent_bom_id) if project.permanent_bom_id else False, } ) @http.route('/api/ifc/documents/bom/create-permanent-if-available', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_bom_create_permanent_if_available(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: permanent_bom = project.action_create_permanent_bom_if_available() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='BOM_PERMANENT_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal membuat Permanent BoM.', code='BOM_PERMANENT_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'permanent_bom': self._serialize_bom(permanent_bom), } ) @http.route('/api/ifc/documents/inventory/check', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_inventory_check(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='read') if error: return error try: availability = project.action_check_inventory_availability() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='INVENTORY_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal cek ketersediaan inventory.', code='INVENTORY_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'availability': availability, } ) @http.route('/api/ifc/documents/inventory/create-draft-purchase', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_inventory_create_draft_purchase(self, document_id=None, vendor_id=None, only_shortage=True, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: purchase_orders = project.action_create_draft_purchase_for_shortage( vendor_id=vendor_id, only_shortage=_to_bool(only_shortage, default=True), ) except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='PURCHASE_DRAFT_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal membuat draft purchase.', code='PURCHASE_DRAFT_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'purchase_orders': [self._serialize_purchase_order(po) for po in purchase_orders], } ) @http.route('/api/ifc/documents/inventory/create-product', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_inventory_create_product(self, document_id=None, ifc_bom_line_id=None, **kwargs): if not document_id or not ifc_bom_line_id: return _jsonrpc_fail('document_id dan ifc_bom_line_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: product, created = project.action_create_missing_product_for_ifc_bom(ifc_bom_line_id) except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='PRODUCT_CREATE_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal membuat produk dari IFC BOM.', code='PRODUCT_CREATE_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'created': created, 'product': { 'id': product.id, 'name': product.display_name, 'default_code': product.default_code, 'qty_available': product.qty_available, 'virtual_available': product.virtual_available, }, } ) @http.route('/api/ifc/documents/mo/create-if-available', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_mo_create_if_available(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: manufacturing_order = project.action_create_mo_if_available() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='MO_CREATE_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal membuat Manufacturing Order.', code='MO_CREATE_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'manufacturing_order': self._serialize_manufacturing_order(manufacturing_order), } ) @http.route('/api/ifc/documents/mo/create-from-permanent-bom', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_mo_create_from_permanent_bom(self, document_id=None, **kwargs): return self.api_ifc_documents_mo_create_if_available(document_id=document_id, **kwargs) @http.route('/api/ifc/documents/mo/check-all-availability', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_mo_check_all_availability(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='read') if error: return error try: summary = project.action_check_all_mo_availability_summary() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='MO_AVAILABILITY_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal cek availability seluruh MO.', code='MO_AVAILABILITY_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'summary': summary, } ) @http.route('/api/ifc/documents/mo/confirm-all-if-available', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_mo_confirm_all_if_available(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: confirmed_mos = project.action_confirm_all_mo_if_available() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='MO_CONFIRM_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal confirm seluruh MO.', code='MO_CONFIRM_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'confirmed_mos': [self._serialize_manufacturing_order(mo) for mo in confirmed_mos], } ) @http.route('/api/ifc/documents/mo/run-pipeline', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_mo_run_pipeline(self, document_id=None, auto_confirm=False, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='write') if error: return error try: pipeline_result = project.action_run_mo_pipeline(auto_confirm=_to_bool(auto_confirm, default=False)) except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='MO_PIPELINE_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal menjalankan pipeline MO.', code='MO_PIPELINE_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'estimated_availability': pipeline_result['estimated_availability'], 'permanent_bom': self._serialize_bom(pipeline_result['permanent_bom']) if pipeline_result['permanent_bom'] else False, 'manufacturing_order': self._serialize_manufacturing_order(pipeline_result['manufacturing_order']) if pipeline_result['manufacturing_order'] else False, 'mo_summary': pipeline_result['mo_summary'], 'auto_confirm': pipeline_result['auto_confirm'], 'confirmed_mos': [self._serialize_manufacturing_order(mo) for mo in pipeline_result['confirmed_mos']], } ) @http.route('/api/ifc/documents/mo/pipeline-dry-run', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_mo_pipeline_dry_run(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project, error = self._get_project_for_action(document_id, mode='read') if error: return error try: preview = project.action_preview_mo_pipeline() except (AccessError, UserError) as exc: return _jsonrpc_fail(str(exc), code='MO_PIPELINE_DRY_RUN_FAILED') except Exception as exc: return _jsonrpc_fail('Gagal menjalankan dry-run pipeline MO.', code='MO_PIPELINE_DRY_RUN_FAILED', details=str(exc)) return _jsonrpc_ok( { 'document': self._serialize_project(project), 'preview': preview, } ) @http.route('/api/ifc/documents/temp-file', type='json', auth='user', methods=['POST'], csrf=False) def api_ifc_documents_temp_file(self, document_id=None, **kwargs): if not document_id: return _jsonrpc_fail('document_id wajib diisi.', code='BAD_REQUEST') project = request.env['grt.ifc.project'].browse(int(document_id)).exists() if not project: return _jsonrpc_fail('Dokumen tidak ditemukan.', code='NOT_FOUND') try: self._check_record_access(project, mode='read') except AccessError: return _jsonrpc_fail('Tidak punya akses membaca file.', code='ACCESS_DENIED') if not project.file_data: return _jsonrpc_fail('Dokumen tidak memiliki file IFC.', code='NO_FILE') self._cleanup_temp_files() token = uuid.uuid4().hex file_name = _safe_filename(project.file_name or ('%s.ifc' % project.document_ref)) temp_root = self._get_temp_root() temp_path = temp_root / ('%s_%s' % (token, file_name)) try: temp_path.write_bytes(base64.b64decode(project.file_data)) except Exception as exc: return _jsonrpc_fail('Gagal decode file IFC ke temp folder.', code='DECODE_FAILED', details=str(exc)) download_url, expires_at = self._build_signed_temp_download_url(temp_path.name, project.id, request.env.user.id) return _jsonrpc_ok( { 'token': token, 'file_name': file_name, 'temp_path': str(temp_path), 'download_url': download_url, 'expires_in_minutes': TEMP_TTL_MINUTES, 'expires_at_unix': expires_at, } ) @http.route('/api/ifc/temp/', type='http', auth='user', methods=['GET'], csrf=False) def api_ifc_temp_get(self, file_key, **kwargs): if not file_key or '/' in file_key or '\\' in file_key: return _json_http_response({'error': 'Invalid file key.'}, status=400) doc_id = kwargs.get('doc_id') signed_uid = kwargs.get('uid') exp = kwargs.get('exp') sig = kwargs.get('sig') if not doc_id or not signed_uid or not exp or not sig: return _json_http_response({'error': 'Signed URL parameters are required.'}, status=400) try: doc_id_int = int(doc_id) signed_uid_int = int(signed_uid) except (TypeError, ValueError): return _json_http_response({'error': 'Invalid signed URL parameters.'}, status=400) if signed_uid_int != request.env.user.id: return _json_http_response({'error': 'Signed URL does not match current user.'}, status=403) if not self._verify_temp_signature(file_key, doc_id_int, signed_uid_int, exp, sig): return _json_http_response({'error': 'Invalid or expired signature.'}, status=403) project = request.env['grt.ifc.project'].browse(doc_id_int).exists() if not project: return _json_http_response({'error': 'Document not found.'}, status=404) try: self._check_record_access(project, mode='read') except AccessError: return _json_http_response({'error': 'Access denied.'}, status=403) temp_path = self._get_temp_root() / file_key if not temp_path.exists() or not temp_path.is_file(): return _json_http_response({'error': 'Temp file not found.'}, status=404) mimetype, _ = mimetypes.guess_type(str(temp_path)) mimetype = mimetype or 'application/octet-stream' content = temp_path.read_bytes() filename = file_key.split('_', 1)[1] if '_' in file_key else file_key headers = [ ('Content-Type', mimetype), ('Content-Length', str(len(content))), ('Content-Disposition', 'inline; filename="%s"' % filename), ] return request.make_response(content, headers=headers)