lyg
2 天以前 22f370322412074174cde20ecfd14ec03657ab63
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
# -*- coding: utf-8 -*-
# @author: lyg
# @date: 2025-5-12
# @version: 1
# @description: 文档数据库模型
 
from sqlalchemy import create_engine, Column, DateTime, Integer, Text, ForeignKey
from sqlalchemy.dialects.mysql import LONGTEXT
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import sessionmaker, scoped_session
 
from knowledgebase.log import Log
 
Base = declarative_base()
metadata = Base.metadata
 
 
class TParagraph(Base):
    """
    段落表
    """
    __tablename__ = 't_paragraphs'
    id = Column(Integer, primary_key=True)
    # text = Column(LONGTEXT)
    text = Column(Text)
    title_level = Column(Integer)
    title_num = Column(Text)
    num_level = Column(Integer)
    num = Column(Integer)
    doc_id = Column(Integer, ForeignKey('t_docs.id'))
    parent_id = Column(Integer, ForeignKey('t_paragraphs.id'))
    parent_link = relationship("TParagraphLink", foreign_keys='TParagraphLink.child_id', back_populates='parent',
                               uselist=False)
    children_links = relationship("TParagraphLink", foreign_keys='TParagraphLink.parent_id', back_populates='child')
    ref_links = relationship("TParagraphRefLink", foreign_keys='TParagraphRefLink.child_id', back_populates='parent')
    entity_links = relationship("TParagraphEntityLink")
    is_del = Column(Integer)
 
    @property
    def children(self):
        return [link.child for link in self.children_links]
 
    @property
    def parent(self):
        if self.parent_link:
            return self.parent_link.parent
        return None
 
    def refs(self):
        return [link.child for link in self.ref_links]
 
 
class TParagraphLink(Base):
    """
    段落link表
    """
    __tablename__ = 't_paragraph_links'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('t_paragraphs.id'))
    parent = relationship("TParagraph", foreign_keys=[parent_id], back_populates="children_links")
    child_id = Column(Integer, ForeignKey('t_paragraphs.id'))
    child = relationship("TParagraph", foreign_keys=[child_id], back_populates="parent_link")
    is_del = Column(Integer)
 
 
class TParagraphRefLink(Base):
    """
    段落引用link表
    """
    __tablename__ = 't_paragraph_ref_links'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('t_paragraphs.id'))
    parent = relationship("TParagraph", foreign_keys=[parent_id], back_populates="ref_links")
    child_id = Column(Integer, ForeignKey('t_paragraphs.id'))
    child = relationship("TParagraph", foreign_keys=[child_id], viewonly=True)
    is_del = Column(Integer)
 
 
class TParagraphEntityLink(Base):
    """
    段落实体link表
    """
    __tablename__ = 't_paragraph_entity_links'
    id = Column(Integer, primary_key=True)
    paragraph_id = Column(Integer, ForeignKey('t_paragraphs.id'))
    paragraph = relationship("TParagraph", foreign_keys=[paragraph_id], back_populates="entity_links")
    entity_id = Column(Integer, ForeignKey('t_entities.id'))
    entity = relationship("TEntity", foreign_keys=[entity_id])
    is_del = Column(Integer)
 
 
class TDoc(Base):
    """
    文档表
    """
    __tablename__ = 't_docs'
    id = Column(Integer, primary_key=True)
    file = Column(Text)
    file_name = Column(Text)
    paragraphs = relationship("TParagraph")
    is_del = Column(Integer)
 
 
class TEntity(Base):
    """
    实体表
    """
    __tablename__ = 't_entities'
    id = Column(Integer, primary_key=True)
    name = Column(Text)
    prompts = Column(Text)
    type = Column(Text)
    doc_type = Column(Text)
 
 
class TEntityLink(Base):
    __tablename__ = 't_entity_links'
    id = Column(Integer, primary_key=True)
    entity_id = Column(Integer, ForeignKey('t_entities.id'))
    is_del = Column(Integer)
 
 
# class TTmPacket(Base):
#     __tablename__ = 't_tm_packets'
#     id = Column(Integer, primary_key=True)
#     name = Column(Text)
#     code = Column(Text)
#     apid = Column(Integer)
#     is_del = Column(Integer)
#
#
# class TTmPacketParagraphLink(Base):
#     __tablename__ = 't_tm_packet_paragraph_links'
#     id = Column(Integer, primary_key=True)
#     tm_packet_id = Column(Integer, ForeignKey('t_tm_packets.id'))
#     paragraph_id = Column(Integer, ForeignKey('t_paragraphs.id'))
#     tm_packet = relationship("TTmPacket", foreign_keys=[tm_packet_id], uselist=False)
#     paragraph = relationship("TParagraph", foreign_keys=[paragraph_id], uselist=False)
#     is_del = Column(Integer)
 
 
def init_doc_db(project_path):
    """
    初始化文档数据库
    :return: 数据库session实例
    """
    # mysql
    Log.info("连接并初始化文档数据库...")
    # engine = create_engine('mysql+pymysql://root:123456@10.74.15.171:3306/knowledgebase_xx25', echo=False)
    engine = create_engine(f'sqlite:///{project_path}/docs/doc_db.db', echo=False)
    Base.metadata.create_all(engine)
    SessionFactory = sessionmaker(bind=engine)
    Session = scoped_session(SessionFactory)
    session = Session()
    return session
 
# _xx25