forked from franciscod/telegram-twitter-forwarder-bot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
models.py
94 lines (74 loc) · 2.95 KB
/
models.py
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
import datetime
import tweepy
from peewee import (Model, DateTimeField, ForeignKeyField, BigIntegerField, CharField,
IntegerField, OperationalError, BooleanField)
from playhouse.migrate import migrate, SqliteMigrator, SqliteDatabase
from tweepy.auth import OAuthHandler
class TwitterUser(Model):
user_id = IntegerField(unique=True)
screen_name = CharField(unique=True)
name = CharField()
@property
def full_name(self):
return "{} ({})".format(self.name, self.screen_name)
class TelegramChat(Model):
chat_id = IntegerField(unique=True)
known_at = DateTimeField(default=datetime.datetime.now)
tg_type = CharField()
last_contact = DateTimeField(default=datetime.datetime.now)
twitter_request_token = CharField(null=True)
twitter_token = CharField(null=True)
twitter_secret = CharField(null=True)
last_tweet_id = BigIntegerField(default=0)
timezone_name = CharField(null=True)
delete_soon = BooleanField(default=False)
@property
def is_group(self):
return self.chat_id < 0
def touch_contact(self):
self.last_contact = datetime.datetime.now()
self.save()
@property
def is_authorized(self):
return self.twitter_token is not None and self.twitter_secret is not None
def tw_api(self, consumer_key, consumer_secret):
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(self.twitter_token, self.twitter_secret)
return tweepy.API(auth)
class Subscription(Model):
tg_chat = ForeignKeyField(TelegramChat, related_name="subscriptions")
tw_user = ForeignKeyField(TwitterUser, related_name="subscriptions")
class Media:
def __init__(self, type, url):
self.type = type
self.url = url
class Tweet:
def __init__(self, *args, **kwargs):
self.id = None
self.text = None
self.created_at = None
self.user_name = None
self.user_screen_name = None
self.link_url = ''
self.media_list = []
for k in kwargs:
setattr(self, k, kwargs[k])
# Create tables
for t in (TwitterUser, TelegramChat, Subscription):
t.create_table(fail_silently=True)
# Migrate new fields. TODO: think of some better migration mechanism
db = SqliteDatabase('peewee.db', timeout=10)
migrator = SqliteMigrator(db)
operations = [
migrator.add_column('telegramchat', 'twitter_request_token', TelegramChat.twitter_request_token),
migrator.add_column('telegramchat', 'twitter_token', TelegramChat.twitter_token),
migrator.add_column('telegramchat', 'twitter_secret', TelegramChat.twitter_secret),
migrator.add_column('telegramchat', 'timezone_name', TelegramChat.timezone_name),
migrator.add_column('telegramchat', 'delete_soon', TelegramChat.delete_soon),
migrator.add_column('telegramchat', 'last_tweet_id', TelegramChat.last_tweet_id),
]
for op in operations:
try:
migrate(op)
except OperationalError:
pass