import config
-def get_score_data(score_db_path, score_id):
+def get_score_data(score_id):
'''DBからスコアデータを取得する。
Args:
- score_db_path: スコアデータが格納されているDBへのパスを表す文字列。
score_id: 取得するスコアのスコアID。
Noneの場合最新のスコアを取得する。
else:
cond = 'WHERE score_id = :score_id'
- with sqlite3.connect(score_db_path) as con:
+ with sqlite3.connect(config.config['ScoreDB']['path']) as con:
con.row_factory = sqlite3.Row
sql = '''
SELECT
return score[0] if len(score) == 1 else None
-def get_daily_score_stats(score_db_path, year, month, day):
+def get_daily_score_stats(year, month, day):
'''DBから指定した日付のスコア統計データを得る
Args:
- score_db_path: スコアデータが格納されているDBへのパスを表す文字列。
year: 指定する年。
month: 指定する月。
day: 指定する日。
取得したスコア統計データを格納した辞書。
'total_count': 総スコア件数, 'winner_count': 勝利スコア件数
'''
- with sqlite3.connect(score_db_path) as con:
+ with sqlite3.connect(config.config['ScoreDB']['path']) as con:
con.row_factory = sqlite3.Row
sql = '''
SELECT
return match.group(1) if match else None
-def create_tweet(score_db_path, score_id):
+def create_tweet(score_id):
'''ツイートするメッセージを生成する。
Args:
- score_db: スコアデータが格納されているDBへのパスを表す文字列。
score_id: 指定するスコアID。Noneの場合最新のスコアを取得する。
Returns:
生成したツイートメッセージ文字列。
なんらかの理由により生成できなかった場合None。
'''
- score_data = get_score_data(score_db_path, options.score_id)
+ score_data = get_score_data(score_id)
if score_data is None:
return None
return tweet
-def create_daily_stats_tweet(score_db_path, year, month, day):
+def create_daily_stats_tweet(year, month, day):
'''デイリースコア統計データのツイートを生成する
Args:
- score_db_path: スコアデータが格納されているDBへのパスを表す文字列。
year: 指定する年。
month: 指定する月。
day: 指定する日。
生成したツイートメッセージ文字列。
なんらかの理由により生成できなかった場合None。
'''
- daily_stats = get_daily_score_stats(score_db_path, year, month, day)
+ daily_stats = get_daily_score_stats(year, month, day)
tweet = (u"{year}年{month}月{day}日のスコア\n"
u"全 {total_count} 件, 勝利 {winner_count} 件\n"
return tweet
-def tweet(oauth, tweet_contents):
+def tweet(tweet_contents):
'''ツイートする。
Args:
from logging import getLogger
logger = getLogger(__name__)
- twitter = OAuth1Session(**oauth)
+ twitter = OAuth1Session(**config.config['TwitterOAuth'])
url = "https://api.twitter.com/1.1/statuses/update.json"
target_datetime = datetime.datetime.strptime(
options.stats_date, "%Y-%m-%d")
tweet_contents = create_daily_stats_tweet(
- config.config['ScoreDB']['path'],
target_datetime.year,
target_datetime.month,
target_datetime.day
)
else:
- tweet_contents = create_tweet(config.config['ScoreDB']['path'],
- options.score_id)
+ tweet_contents = create_tweet(options.score_id)
if tweet_contents is None:
logger.warning('No score data found.')
if (options.dry_run):
print(tweet_contents.encode("UTF-8"))
else:
- tweet(config.config['TwitterOAuth'], tweet_contents)
+ tweet(tweet_contents)
except Exception:
from traceback import format_exc
logger.critical(format_exc())