datasets_document.py 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874
  1. # -*- coding:utf-8 -*-
  2. import random
  3. from datetime import datetime
  4. from typing import List
  5. from flask import request
  6. from flask_login import login_required, current_user
  7. from flask_restful import Resource, fields, marshal, marshal_with, reqparse
  8. from sqlalchemy import desc, asc
  9. from werkzeug.exceptions import NotFound, Forbidden
  10. import services
  11. from controllers.console import api
  12. from controllers.console.app.error import ProviderNotInitializeError, ProviderQuotaExceededError, \
  13. ProviderModelCurrentlyNotSupportError
  14. from controllers.console.datasets.error import DocumentAlreadyFinishedError, InvalidActionError, DocumentIndexingError, \
  15. InvalidMetadataError, ArchivedDocumentImmutableError
  16. from controllers.console.setup import setup_required
  17. from controllers.console.wraps import account_initialization_required
  18. from core.indexing_runner import IndexingRunner
  19. from core.llm.error import ProviderTokenNotInitError, QuotaExceededError, ModelCurrentlyNotSupportError
  20. from extensions.ext_redis import redis_client
  21. from libs.helper import TimestampField
  22. from extensions.ext_database import db
  23. from models.dataset import DatasetProcessRule, Dataset
  24. from models.dataset import Document, DocumentSegment
  25. from models.model import UploadFile
  26. from services.dataset_service import DocumentService, DatasetService
  27. from tasks.add_document_to_index_task import add_document_to_index_task
  28. from tasks.remove_document_from_index_task import remove_document_from_index_task
  29. dataset_fields = {
  30. 'id': fields.String,
  31. 'name': fields.String,
  32. 'description': fields.String,
  33. 'permission': fields.String,
  34. 'data_source_type': fields.String,
  35. 'indexing_technique': fields.String,
  36. 'created_by': fields.String,
  37. 'created_at': TimestampField,
  38. }
  39. document_fields = {
  40. 'id': fields.String,
  41. 'position': fields.Integer,
  42. 'data_source_type': fields.String,
  43. 'data_source_info': fields.Raw(attribute='data_source_info_dict'),
  44. 'dataset_process_rule_id': fields.String,
  45. 'name': fields.String,
  46. 'created_from': fields.String,
  47. 'created_by': fields.String,
  48. 'created_at': TimestampField,
  49. 'tokens': fields.Integer,
  50. 'indexing_status': fields.String,
  51. 'error': fields.String,
  52. 'enabled': fields.Boolean,
  53. 'disabled_at': TimestampField,
  54. 'disabled_by': fields.String,
  55. 'archived': fields.Boolean,
  56. 'display_status': fields.String,
  57. 'word_count': fields.Integer,
  58. 'hit_count': fields.Integer,
  59. }
  60. document_with_segments_fields = {
  61. 'id': fields.String,
  62. 'position': fields.Integer,
  63. 'data_source_type': fields.String,
  64. 'data_source_info': fields.Raw(attribute='data_source_info_dict'),
  65. 'dataset_process_rule_id': fields.String,
  66. 'name': fields.String,
  67. 'created_from': fields.String,
  68. 'created_by': fields.String,
  69. 'created_at': TimestampField,
  70. 'tokens': fields.Integer,
  71. 'indexing_status': fields.String,
  72. 'error': fields.String,
  73. 'enabled': fields.Boolean,
  74. 'disabled_at': TimestampField,
  75. 'disabled_by': fields.String,
  76. 'archived': fields.Boolean,
  77. 'display_status': fields.String,
  78. 'word_count': fields.Integer,
  79. 'hit_count': fields.Integer,
  80. 'completed_segments': fields.Integer,
  81. 'total_segments': fields.Integer
  82. }
  83. class DocumentResource(Resource):
  84. def get_document(self, dataset_id: str, document_id: str) -> Document:
  85. dataset = DatasetService.get_dataset(dataset_id)
  86. if not dataset:
  87. raise NotFound('Dataset not found.')
  88. try:
  89. DatasetService.check_dataset_permission(dataset, current_user)
  90. except services.errors.account.NoPermissionError as e:
  91. raise Forbidden(str(e))
  92. document = DocumentService.get_document(dataset_id, document_id)
  93. if not document:
  94. raise NotFound('Document not found.')
  95. if document.tenant_id != current_user.current_tenant_id:
  96. raise Forbidden('No permission.')
  97. return document
  98. def get_batch_documents(self, dataset_id: str, batch: str) -> List[Document]:
  99. dataset = DatasetService.get_dataset(dataset_id)
  100. if not dataset:
  101. raise NotFound('Dataset not found.')
  102. try:
  103. DatasetService.check_dataset_permission(dataset, current_user)
  104. except services.errors.account.NoPermissionError as e:
  105. raise Forbidden(str(e))
  106. documents = DocumentService.get_batch_documents(dataset_id, batch)
  107. if not documents:
  108. raise NotFound('Documents not found.')
  109. return documents
  110. class GetProcessRuleApi(Resource):
  111. @setup_required
  112. @login_required
  113. @account_initialization_required
  114. def get(self):
  115. req_data = request.args
  116. document_id = req_data.get('document_id')
  117. if document_id:
  118. # get the latest process rule
  119. document = Document.query.get_or_404(document_id)
  120. dataset = DatasetService.get_dataset(document.dataset_id)
  121. if not dataset:
  122. raise NotFound('Dataset not found.')
  123. try:
  124. DatasetService.check_dataset_permission(dataset, current_user)
  125. except services.errors.account.NoPermissionError as e:
  126. raise Forbidden(str(e))
  127. # get the latest process rule
  128. dataset_process_rule = db.session.query(DatasetProcessRule). \
  129. filter(DatasetProcessRule.dataset_id == document.dataset_id). \
  130. order_by(DatasetProcessRule.created_at.desc()). \
  131. limit(1). \
  132. one_or_none()
  133. mode = dataset_process_rule.mode
  134. rules = dataset_process_rule.rules_dict
  135. else:
  136. mode = DocumentService.DEFAULT_RULES['mode']
  137. rules = DocumentService.DEFAULT_RULES['rules']
  138. return {
  139. 'mode': mode,
  140. 'rules': rules
  141. }
  142. class DatasetDocumentListApi(Resource):
  143. @setup_required
  144. @login_required
  145. @account_initialization_required
  146. def get(self, dataset_id):
  147. dataset_id = str(dataset_id)
  148. page = request.args.get('page', default=1, type=int)
  149. limit = request.args.get('limit', default=20, type=int)
  150. search = request.args.get('keyword', default=None, type=str)
  151. sort = request.args.get('sort', default='-created_at', type=str)
  152. fetch = request.args.get('fetch', default=False, type=bool)
  153. dataset = DatasetService.get_dataset(dataset_id)
  154. if not dataset:
  155. raise NotFound('Dataset not found.')
  156. try:
  157. DatasetService.check_dataset_permission(dataset, current_user)
  158. except services.errors.account.NoPermissionError as e:
  159. raise Forbidden(str(e))
  160. query = Document.query.filter_by(
  161. dataset_id=str(dataset_id), tenant_id=current_user.current_tenant_id)
  162. if search:
  163. search = f'%{search}%'
  164. query = query.filter(Document.name.like(search))
  165. if sort.startswith('-'):
  166. sort_logic = desc
  167. sort = sort[1:]
  168. else:
  169. sort_logic = asc
  170. if sort == 'hit_count':
  171. sub_query = db.select(DocumentSegment.document_id,
  172. db.func.sum(DocumentSegment.hit_count).label("total_hit_count")) \
  173. .group_by(DocumentSegment.document_id) \
  174. .subquery()
  175. query = query.outerjoin(sub_query, sub_query.c.document_id == Document.id) \
  176. .order_by(sort_logic(db.func.coalesce(sub_query.c.total_hit_count, 0)))
  177. elif sort == 'created_at':
  178. query = query.order_by(sort_logic(Document.created_at))
  179. else:
  180. query = query.order_by(desc(Document.created_at))
  181. paginated_documents = query.paginate(
  182. page=page, per_page=limit, max_per_page=100, error_out=False)
  183. documents = paginated_documents.items
  184. if fetch:
  185. for document in documents:
  186. completed_segments = DocumentSegment.query.filter(DocumentSegment.completed_at.isnot(None),
  187. DocumentSegment.document_id == str(document.id),
  188. DocumentSegment.status != 're_segment').count()
  189. total_segments = DocumentSegment.query.filter(DocumentSegment.document_id == str(document.id),
  190. DocumentSegment.status != 're_segment').count()
  191. document.completed_segments = completed_segments
  192. document.total_segments = total_segments
  193. data = marshal(documents, document_with_segments_fields)
  194. else:
  195. data = marshal(documents, document_fields)
  196. response = {
  197. 'data': data,
  198. 'has_more': len(documents) == limit,
  199. 'limit': limit,
  200. 'total': paginated_documents.total,
  201. 'page': page
  202. }
  203. return response
  204. documents_and_batch_fields = {
  205. 'documents': fields.List(fields.Nested(document_fields)),
  206. 'batch': fields.String
  207. }
  208. @setup_required
  209. @login_required
  210. @account_initialization_required
  211. @marshal_with(documents_and_batch_fields)
  212. def post(self, dataset_id):
  213. dataset_id = str(dataset_id)
  214. dataset = DatasetService.get_dataset(dataset_id)
  215. if not dataset:
  216. raise NotFound('Dataset not found.')
  217. # The role of the current user in the ta table must be admin or owner
  218. if current_user.current_tenant.current_role not in ['admin', 'owner']:
  219. raise Forbidden()
  220. try:
  221. DatasetService.check_dataset_permission(dataset, current_user)
  222. except services.errors.account.NoPermissionError as e:
  223. raise Forbidden(str(e))
  224. parser = reqparse.RequestParser()
  225. parser.add_argument('indexing_technique', type=str, choices=Dataset.INDEXING_TECHNIQUE_LIST, nullable=False,
  226. location='json')
  227. parser.add_argument('data_source', type=dict, required=False, location='json')
  228. parser.add_argument('process_rule', type=dict, required=False, location='json')
  229. parser.add_argument('duplicate', type=bool, nullable=False, location='json')
  230. parser.add_argument('original_document_id', type=str, required=False, location='json')
  231. args = parser.parse_args()
  232. if not dataset.indexing_technique and not args['indexing_technique']:
  233. raise ValueError('indexing_technique is required.')
  234. # validate args
  235. DocumentService.document_create_args_validate(args)
  236. try:
  237. documents, batch = DocumentService.save_document_with_dataset_id(dataset, args, current_user)
  238. except ProviderTokenNotInitError:
  239. raise ProviderNotInitializeError()
  240. except QuotaExceededError:
  241. raise ProviderQuotaExceededError()
  242. except ModelCurrentlyNotSupportError:
  243. raise ProviderModelCurrentlyNotSupportError()
  244. return {
  245. 'documents': documents,
  246. 'batch': batch
  247. }
  248. class DatasetInitApi(Resource):
  249. dataset_and_document_fields = {
  250. 'dataset': fields.Nested(dataset_fields),
  251. 'documents': fields.List(fields.Nested(document_fields)),
  252. 'batch': fields.String
  253. }
  254. @setup_required
  255. @login_required
  256. @account_initialization_required
  257. @marshal_with(dataset_and_document_fields)
  258. def post(self):
  259. # The role of the current user in the ta table must be admin or owner
  260. if current_user.current_tenant.current_role not in ['admin', 'owner']:
  261. raise Forbidden()
  262. parser = reqparse.RequestParser()
  263. parser.add_argument('indexing_technique', type=str, choices=Dataset.INDEXING_TECHNIQUE_LIST, required=True,
  264. nullable=False, location='json')
  265. parser.add_argument('data_source', type=dict, required=True, nullable=True, location='json')
  266. parser.add_argument('process_rule', type=dict, required=True, nullable=True, location='json')
  267. args = parser.parse_args()
  268. # validate args
  269. DocumentService.document_create_args_validate(args)
  270. try:
  271. dataset, documents, batch = DocumentService.save_document_without_dataset_id(
  272. tenant_id=current_user.current_tenant_id,
  273. document_data=args,
  274. account=current_user
  275. )
  276. except ProviderTokenNotInitError:
  277. raise ProviderNotInitializeError()
  278. except QuotaExceededError:
  279. raise ProviderQuotaExceededError()
  280. except ModelCurrentlyNotSupportError:
  281. raise ProviderModelCurrentlyNotSupportError()
  282. response = {
  283. 'dataset': dataset,
  284. 'documents': documents,
  285. 'batch': batch
  286. }
  287. return response
  288. class DocumentIndexingEstimateApi(DocumentResource):
  289. @setup_required
  290. @login_required
  291. @account_initialization_required
  292. def get(self, dataset_id, document_id):
  293. dataset_id = str(dataset_id)
  294. document_id = str(document_id)
  295. document = self.get_document(dataset_id, document_id)
  296. if document.indexing_status in ['completed', 'error']:
  297. raise DocumentAlreadyFinishedError()
  298. data_process_rule = document.dataset_process_rule
  299. data_process_rule_dict = data_process_rule.to_dict()
  300. response = {
  301. "tokens": 0,
  302. "total_price": 0,
  303. "currency": "USD",
  304. "total_segments": 0,
  305. "preview": []
  306. }
  307. if document.data_source_type == 'upload_file':
  308. data_source_info = document.data_source_info_dict
  309. if data_source_info and 'upload_file_id' in data_source_info:
  310. file_id = data_source_info['upload_file_id']
  311. file = db.session.query(UploadFile).filter(
  312. UploadFile.tenant_id == document.tenant_id,
  313. UploadFile.id == file_id
  314. ).first()
  315. # raise error if file not found
  316. if not file:
  317. raise NotFound('File not found.')
  318. indexing_runner = IndexingRunner()
  319. response = indexing_runner.file_indexing_estimate([file], data_process_rule_dict)
  320. return response
  321. class DocumentBatchIndexingEstimateApi(DocumentResource):
  322. @setup_required
  323. @login_required
  324. @account_initialization_required
  325. def get(self, dataset_id, batch):
  326. dataset_id = str(dataset_id)
  327. batch = str(batch)
  328. dataset = DatasetService.get_dataset(dataset_id)
  329. if dataset is None:
  330. raise NotFound("Dataset not found.")
  331. documents = self.get_batch_documents(dataset_id, batch)
  332. response = {
  333. "tokens": 0,
  334. "total_price": 0,
  335. "currency": "USD",
  336. "total_segments": 0,
  337. "preview": []
  338. }
  339. if not documents:
  340. return response
  341. data_process_rule = documents[0].dataset_process_rule
  342. data_process_rule_dict = data_process_rule.to_dict()
  343. info_list = []
  344. for document in documents:
  345. if document.indexing_status in ['completed', 'error']:
  346. raise DocumentAlreadyFinishedError()
  347. data_source_info = document.data_source_info_dict
  348. # format document files info
  349. if data_source_info and 'upload_file_id' in data_source_info:
  350. file_id = data_source_info['upload_file_id']
  351. info_list.append(file_id)
  352. # format document notion info
  353. elif data_source_info and 'notion_workspace_id' in data_source_info and 'notion_page_id' in data_source_info:
  354. pages = []
  355. page = {
  356. 'page_id': data_source_info['notion_page_id'],
  357. 'type': data_source_info['type']
  358. }
  359. pages.append(page)
  360. notion_info = {
  361. 'workspace_id': data_source_info['notion_workspace_id'],
  362. 'pages': pages
  363. }
  364. info_list.append(notion_info)
  365. if dataset.data_source_type == 'upload_file':
  366. file_details = db.session.query(UploadFile).filter(
  367. UploadFile.tenant_id == current_user.current_tenant_id,
  368. UploadFile.id in info_list
  369. ).all()
  370. if file_details is None:
  371. raise NotFound("File not found.")
  372. indexing_runner = IndexingRunner()
  373. response = indexing_runner.file_indexing_estimate(file_details, data_process_rule_dict)
  374. elif dataset.data_source_type:
  375. indexing_runner = IndexingRunner()
  376. response = indexing_runner.notion_indexing_estimate(info_list,
  377. data_process_rule_dict)
  378. else:
  379. raise ValueError('Data source type not support')
  380. return response
  381. class DocumentBatchIndexingStatusApi(DocumentResource):
  382. document_status_fields = {
  383. 'id': fields.String,
  384. 'indexing_status': fields.String,
  385. 'processing_started_at': TimestampField,
  386. 'parsing_completed_at': TimestampField,
  387. 'cleaning_completed_at': TimestampField,
  388. 'splitting_completed_at': TimestampField,
  389. 'completed_at': TimestampField,
  390. 'paused_at': TimestampField,
  391. 'error': fields.String,
  392. 'stopped_at': TimestampField,
  393. 'completed_segments': fields.Integer,
  394. 'total_segments': fields.Integer,
  395. }
  396. document_status_fields_list = {
  397. 'data': fields.List(fields.Nested(document_status_fields))
  398. }
  399. @setup_required
  400. @login_required
  401. @account_initialization_required
  402. def get(self, dataset_id, batch):
  403. dataset_id = str(dataset_id)
  404. batch = str(batch)
  405. documents = self.get_batch_documents(dataset_id, batch)
  406. documents_status = []
  407. for document in documents:
  408. completed_segments = DocumentSegment.query.filter(DocumentSegment.completed_at.isnot(None),
  409. DocumentSegment.document_id == str(document.id),
  410. DocumentSegment.status != 're_segment').count()
  411. total_segments = DocumentSegment.query.filter(DocumentSegment.document_id == str(document.id),
  412. DocumentSegment.status != 're_segment').count()
  413. document.completed_segments = completed_segments
  414. document.total_segments = total_segments
  415. documents_status.append(marshal(document, self.document_status_fields))
  416. data = {
  417. 'data': documents_status
  418. }
  419. return data
  420. class DocumentIndexingStatusApi(DocumentResource):
  421. document_status_fields = {
  422. 'id': fields.String,
  423. 'indexing_status': fields.String,
  424. 'processing_started_at': TimestampField,
  425. 'parsing_completed_at': TimestampField,
  426. 'cleaning_completed_at': TimestampField,
  427. 'splitting_completed_at': TimestampField,
  428. 'completed_at': TimestampField,
  429. 'paused_at': TimestampField,
  430. 'error': fields.String,
  431. 'stopped_at': TimestampField,
  432. 'completed_segments': fields.Integer,
  433. 'total_segments': fields.Integer,
  434. }
  435. @setup_required
  436. @login_required
  437. @account_initialization_required
  438. def get(self, dataset_id, document_id):
  439. dataset_id = str(dataset_id)
  440. document_id = str(document_id)
  441. document = self.get_document(dataset_id, document_id)
  442. completed_segments = DocumentSegment.query \
  443. .filter(DocumentSegment.completed_at.isnot(None),
  444. DocumentSegment.document_id == str(document_id),
  445. DocumentSegment.status != 're_segment') \
  446. .count()
  447. total_segments = DocumentSegment.query \
  448. .filter(DocumentSegment.document_id == str(document_id),
  449. DocumentSegment.status != 're_segment') \
  450. .count()
  451. document.completed_segments = completed_segments
  452. document.total_segments = total_segments
  453. return marshal(document, self.document_status_fields)
  454. class DocumentDetailApi(DocumentResource):
  455. METADATA_CHOICES = {'all', 'only', 'without'}
  456. @setup_required
  457. @login_required
  458. @account_initialization_required
  459. def get(self, dataset_id, document_id):
  460. dataset_id = str(dataset_id)
  461. document_id = str(document_id)
  462. document = self.get_document(dataset_id, document_id)
  463. metadata = request.args.get('metadata', 'all')
  464. if metadata not in self.METADATA_CHOICES:
  465. raise InvalidMetadataError(f'Invalid metadata value: {metadata}')
  466. if metadata == 'only':
  467. response = {
  468. 'id': document.id,
  469. 'doc_type': document.doc_type,
  470. 'doc_metadata': document.doc_metadata
  471. }
  472. elif metadata == 'without':
  473. process_rules = DatasetService.get_process_rules(dataset_id)
  474. data_source_info = document.data_source_detail_dict
  475. response = {
  476. 'id': document.id,
  477. 'position': document.position,
  478. 'data_source_type': document.data_source_type,
  479. 'data_source_info': data_source_info,
  480. 'dataset_process_rule_id': document.dataset_process_rule_id,
  481. 'dataset_process_rule': process_rules,
  482. 'name': document.name,
  483. 'created_from': document.created_from,
  484. 'created_by': document.created_by,
  485. 'created_at': document.created_at.timestamp(),
  486. 'tokens': document.tokens,
  487. 'indexing_status': document.indexing_status,
  488. 'completed_at': int(document.completed_at.timestamp()) if document.completed_at else None,
  489. 'updated_at': int(document.updated_at.timestamp()) if document.updated_at else None,
  490. 'indexing_latency': document.indexing_latency,
  491. 'error': document.error,
  492. 'enabled': document.enabled,
  493. 'disabled_at': int(document.disabled_at.timestamp()) if document.disabled_at else None,
  494. 'disabled_by': document.disabled_by,
  495. 'archived': document.archived,
  496. 'segment_count': document.segment_count,
  497. 'average_segment_length': document.average_segment_length,
  498. 'hit_count': document.hit_count,
  499. 'display_status': document.display_status
  500. }
  501. else:
  502. process_rules = DatasetService.get_process_rules(dataset_id)
  503. data_source_info = document.data_source_detail_dict_()
  504. response = {
  505. 'id': document.id,
  506. 'position': document.position,
  507. 'data_source_type': document.data_source_type,
  508. 'data_source_info': data_source_info,
  509. 'dataset_process_rule_id': document.dataset_process_rule_id,
  510. 'dataset_process_rule': process_rules,
  511. 'name': document.name,
  512. 'created_from': document.created_from,
  513. 'created_by': document.created_by,
  514. 'created_at': document.created_at.timestamp(),
  515. 'tokens': document.tokens,
  516. 'indexing_status': document.indexing_status,
  517. 'completed_at': int(document.completed_at.timestamp()) if document.completed_at else None,
  518. 'updated_at': int(document.updated_at.timestamp()) if document.updated_at else None,
  519. 'indexing_latency': document.indexing_latency,
  520. 'error': document.error,
  521. 'enabled': document.enabled,
  522. 'disabled_at': int(document.disabled_at.timestamp()) if document.disabled_at else None,
  523. 'disabled_by': document.disabled_by,
  524. 'archived': document.archived,
  525. 'doc_type': document.doc_type,
  526. 'doc_metadata': document.doc_metadata,
  527. 'segment_count': document.segment_count,
  528. 'average_segment_length': document.average_segment_length,
  529. 'hit_count': document.hit_count,
  530. 'display_status': document.display_status
  531. }
  532. return response, 200
  533. class DocumentProcessingApi(DocumentResource):
  534. @setup_required
  535. @login_required
  536. @account_initialization_required
  537. def patch(self, dataset_id, document_id, action):
  538. dataset_id = str(dataset_id)
  539. document_id = str(document_id)
  540. document = self.get_document(dataset_id, document_id)
  541. # The role of the current user in the ta table must be admin or owner
  542. if current_user.current_tenant.current_role not in ['admin', 'owner']:
  543. raise Forbidden()
  544. if action == "pause":
  545. if document.indexing_status != "indexing":
  546. raise InvalidActionError('Document not in indexing state.')
  547. document.paused_by = current_user.id
  548. document.paused_at = datetime.utcnow()
  549. document.is_paused = True
  550. db.session.commit()
  551. elif action == "resume":
  552. if document.indexing_status not in ["paused", "error"]:
  553. raise InvalidActionError('Document not in paused or error state.')
  554. document.paused_by = None
  555. document.paused_at = None
  556. document.is_paused = False
  557. db.session.commit()
  558. else:
  559. raise InvalidActionError()
  560. return {'result': 'success'}, 200
  561. class DocumentDeleteApi(DocumentResource):
  562. @setup_required
  563. @login_required
  564. @account_initialization_required
  565. def delete(self, dataset_id, document_id):
  566. dataset_id = str(dataset_id)
  567. document_id = str(document_id)
  568. document = self.get_document(dataset_id, document_id)
  569. try:
  570. DocumentService.delete_document(document)
  571. except services.errors.document.DocumentIndexingError:
  572. raise DocumentIndexingError('Cannot delete document during indexing.')
  573. return {'result': 'success'}, 204
  574. class DocumentMetadataApi(DocumentResource):
  575. @setup_required
  576. @login_required
  577. @account_initialization_required
  578. def put(self, dataset_id, document_id):
  579. dataset_id = str(dataset_id)
  580. document_id = str(document_id)
  581. document = self.get_document(dataset_id, document_id)
  582. req_data = request.get_json()
  583. doc_type = req_data.get('doc_type')
  584. doc_metadata = req_data.get('doc_metadata')
  585. # The role of the current user in the ta table must be admin or owner
  586. if current_user.current_tenant.current_role not in ['admin', 'owner']:
  587. raise Forbidden()
  588. if doc_type is None or doc_metadata is None:
  589. raise ValueError('Both doc_type and doc_metadata must be provided.')
  590. if doc_type not in DocumentService.DOCUMENT_METADATA_SCHEMA:
  591. raise ValueError('Invalid doc_type.')
  592. if not isinstance(doc_metadata, dict):
  593. raise ValueError('doc_metadata must be a dictionary.')
  594. metadata_schema = DocumentService.DOCUMENT_METADATA_SCHEMA[doc_type]
  595. document.doc_metadata = {}
  596. for key, value_type in metadata_schema.items():
  597. value = doc_metadata.get(key)
  598. if value is not None and isinstance(value, value_type):
  599. document.doc_metadata[key] = value
  600. document.doc_type = doc_type
  601. document.updated_at = datetime.utcnow()
  602. db.session.commit()
  603. return {'result': 'success', 'message': 'Document metadata updated.'}, 200
  604. class DocumentStatusApi(DocumentResource):
  605. @setup_required
  606. @login_required
  607. @account_initialization_required
  608. def patch(self, dataset_id, document_id, action):
  609. dataset_id = str(dataset_id)
  610. document_id = str(document_id)
  611. document = self.get_document(dataset_id, document_id)
  612. # The role of the current user in the ta table must be admin or owner
  613. if current_user.current_tenant.current_role not in ['admin', 'owner']:
  614. raise Forbidden()
  615. indexing_cache_key = 'document_{}_indexing'.format(document.id)
  616. cache_result = redis_client.get(indexing_cache_key)
  617. if cache_result is not None:
  618. raise InvalidActionError("Document is being indexed, please try again later")
  619. if action == "enable":
  620. if document.enabled:
  621. raise InvalidActionError('Document already enabled.')
  622. document.enabled = True
  623. document.disabled_at = None
  624. document.disabled_by = None
  625. document.updated_at = datetime.utcnow()
  626. db.session.commit()
  627. # Set cache to prevent indexing the same document multiple times
  628. redis_client.setex(indexing_cache_key, 600, 1)
  629. add_document_to_index_task.delay(document_id)
  630. return {'result': 'success'}, 200
  631. elif action == "disable":
  632. if not document.completed_at or document.indexing_status != 'completed':
  633. raise InvalidActionError('Document is not completed.')
  634. if not document.enabled:
  635. raise InvalidActionError('Document already disabled.')
  636. document.enabled = False
  637. document.disabled_at = datetime.utcnow()
  638. document.disabled_by = current_user.id
  639. document.updated_at = datetime.utcnow()
  640. db.session.commit()
  641. # Set cache to prevent indexing the same document multiple times
  642. redis_client.setex(indexing_cache_key, 600, 1)
  643. remove_document_from_index_task.delay(document_id)
  644. return {'result': 'success'}, 200
  645. elif action == "archive":
  646. if document.archived:
  647. raise InvalidActionError('Document already archived.')
  648. document.archived = True
  649. document.archived_at = datetime.utcnow()
  650. document.archived_by = current_user.id
  651. document.updated_at = datetime.utcnow()
  652. db.session.commit()
  653. if document.enabled:
  654. # Set cache to prevent indexing the same document multiple times
  655. redis_client.setex(indexing_cache_key, 600, 1)
  656. remove_document_from_index_task.delay(document_id)
  657. return {'result': 'success'}, 200
  658. else:
  659. raise InvalidActionError()
  660. class DocumentPauseApi(DocumentResource):
  661. def patch(self, dataset_id, document_id):
  662. """pause document."""
  663. dataset_id = str(dataset_id)
  664. document_id = str(document_id)
  665. dataset = DatasetService.get_dataset(dataset_id)
  666. if not dataset:
  667. raise NotFound('Dataset not found.')
  668. document = DocumentService.get_document(dataset.id, document_id)
  669. # 404 if document not found
  670. if document is None:
  671. raise NotFound("Document Not Exists.")
  672. # 403 if document is archived
  673. if DocumentService.check_archived(document):
  674. raise ArchivedDocumentImmutableError()
  675. try:
  676. # pause document
  677. DocumentService.pause_document(document)
  678. except services.errors.document.DocumentIndexingError:
  679. raise DocumentIndexingError('Cannot pause completed document.')
  680. return {'result': 'success'}, 204
  681. class DocumentRecoverApi(DocumentResource):
  682. def patch(self, dataset_id, document_id):
  683. """recover document."""
  684. dataset_id = str(dataset_id)
  685. document_id = str(document_id)
  686. dataset = DatasetService.get_dataset(dataset_id)
  687. if not dataset:
  688. raise NotFound('Dataset not found.')
  689. document = DocumentService.get_document(dataset.id, document_id)
  690. # 404 if document not found
  691. if document is None:
  692. raise NotFound("Document Not Exists.")
  693. # 403 if document is archived
  694. if DocumentService.check_archived(document):
  695. raise ArchivedDocumentImmutableError()
  696. try:
  697. # pause document
  698. DocumentService.recover_document(document)
  699. except services.errors.document.DocumentIndexingError:
  700. raise DocumentIndexingError('Document is not in paused status.')
  701. return {'result': 'success'}, 204
  702. api.add_resource(GetProcessRuleApi, '/datasets/process-rule')
  703. api.add_resource(DatasetDocumentListApi,
  704. '/datasets/<uuid:dataset_id>/documents')
  705. api.add_resource(DatasetInitApi,
  706. '/datasets/init')
  707. api.add_resource(DocumentIndexingEstimateApi,
  708. '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/indexing-estimate')
  709. api.add_resource(DocumentBatchIndexingEstimateApi,
  710. '/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-estimate')
  711. api.add_resource(DocumentBatchIndexingStatusApi,
  712. '/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-status')
  713. api.add_resource(DocumentIndexingStatusApi,
  714. '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/indexing-status')
  715. api.add_resource(DocumentDetailApi,
  716. '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>')
  717. api.add_resource(DocumentProcessingApi,
  718. '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/<string:action>')
  719. api.add_resource(DocumentDeleteApi,
  720. '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>')
  721. api.add_resource(DocumentMetadataApi,
  722. '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/metadata')
  723. api.add_resource(DocumentStatusApi,
  724. '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/status/<string:action>')
  725. api.add_resource(DocumentPauseApi, '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/pause')
  726. api.add_resource(DocumentRecoverApi, '/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/resume')