このコミットが含まれているのは:
@@ -0,0 +1,98 @@
|
||||
class CountsAxes
|
||||
GRAINS = ['year', 'quarter', 'month', 'week', 'day', 'hour'].freeze
|
||||
|
||||
def self.time_expression(expression, grain, timezone)
|
||||
raise ArgumentError unless GRAINS.include?(grain)
|
||||
|
||||
connection = ApplicationRecord.connection
|
||||
# Use the installed TZInfo database, including DST, rather than MySQL timezone tables.
|
||||
first = Time.utc(1900)
|
||||
last = Time.utc(2101)
|
||||
offset = timezone.period_for_utc(first).utc_total_offset
|
||||
clauses = timezone.transitions_up_to(last, first).filter_map do |transition|
|
||||
next if transition.offset.utc_total_offset == offset
|
||||
|
||||
previous = offset
|
||||
offset = transition.offset.utc_total_offset
|
||||
"WHEN #{ expression } < #{ connection.quote(transition.at.to_time) } THEN #{ previous }"
|
||||
end
|
||||
seconds = clauses.empty? ? offset.to_s : "CASE #{ clauses.join(' ') } ELSE #{ offset } END"
|
||||
local = "DATE_ADD(#{ expression }, INTERVAL (#{ seconds }) SECOND)"
|
||||
case grain
|
||||
when 'year' then "DATE_FORMAT(#{ local }, '%Y')"
|
||||
when 'quarter' then "CONCAT(YEAR(#{ local }), '-Q', QUARTER(#{ local }))"
|
||||
when 'month' then "DATE_FORMAT(#{ local }, '%Y-%m')"
|
||||
when 'week' then "DATE_FORMAT(#{ local }, '%x-W%v')"
|
||||
when 'day' then "DATE_FORMAT(#{ local }, '%Y-%m-%d')"
|
||||
else "DATE_FORMAT(#{ local }, '%Y-%m-%d %H:00')"
|
||||
end
|
||||
end
|
||||
|
||||
attr_reader :rows, :columns, :all
|
||||
|
||||
def initialize(population)
|
||||
@population = population
|
||||
@definition = population.definition
|
||||
@rows = parse(population.params.fetch('rows', '[]'))
|
||||
@columns = parse(population.params.fetch('cols', '[]'))
|
||||
@all = rows + columns
|
||||
raise ArgumentError if all.length > 8 || all.map { _1['field'] }.uniq.length != all.length
|
||||
end
|
||||
|
||||
def expressions
|
||||
all.map do |axis|
|
||||
field = axis.fetch('field')
|
||||
case field
|
||||
when 'tags'
|
||||
"CASE WHEN tag_entry.tag_id IS NOT NULL " +
|
||||
"THEN CONCAT('internal:', tag_entry.tag_id) " +
|
||||
"ELSE CONCAT('external:', tag_entry.external_tag_id) END"
|
||||
when 'changed_field' then 'changed.field_key'
|
||||
else
|
||||
expression = @definition.expression(field)
|
||||
@definition.fields[field] == :datetime ? self.class.time_expression(
|
||||
expression, axis.fetch('grain', 'month'), @population.timezone) : expression
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def joins
|
||||
result = ''
|
||||
if all.any? { _1['field'] == 'tags' }
|
||||
result += <<~SQL.squish
|
||||
JOIN JSON_TABLE(CAST(v.tags_json AS JSON), '$[*]' COLUMNS (
|
||||
tag_id BIGINT PATH '$.tag_id' NULL ON EMPTY,
|
||||
external_tag_id BIGINT PATH '$.external_tag_id' NULL ON EMPTY)) tag_entry
|
||||
|
||||
SQL
|
||||
end
|
||||
if all.any? { _1['field'] == 'changed_field' }
|
||||
fields = @definition.changes.map do |field|
|
||||
"SELECT #{ ApplicationRecord.connection.quote(field) } AS field_key"
|
||||
end.join(' UNION ALL ')
|
||||
condition = @definition.changes.map do |field|
|
||||
"(changed.field_key = '#{ field }' AND #{ @population.changed_sql(field) })"
|
||||
end.join(' OR ')
|
||||
result += " JOIN (#{ fields }) changed ON (#{ condition })"
|
||||
end
|
||||
result
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def parse(value)
|
||||
axes = JSON.parse(value)
|
||||
raise ArgumentError unless axes.is_a?(Array)
|
||||
|
||||
axes.each do |axis|
|
||||
raise ArgumentError unless axis.is_a?(Hash) && (axis.keys - ['field', 'grain']).empty?
|
||||
raise ArgumentError unless @definition.axes(@population.history).include?(axis['field'])
|
||||
if @definition.fields[axis['field']] == :datetime
|
||||
raise ArgumentError unless GRAINS.include?(axis.fetch('grain', 'month'))
|
||||
elsif axis.key?('grain')
|
||||
raise ArgumentError
|
||||
end
|
||||
end
|
||||
axes
|
||||
end
|
||||
end
|
||||
@@ -2,7 +2,7 @@ require 'set'
|
||||
|
||||
# Export identifiers are stable; UI translations are supplied separately as column labels.
|
||||
class CountsExportProjection
|
||||
ROOT_KEYS = { 'resource_id' => 'resource_id', 'version_no' => 'version_no',
|
||||
ROOT_KEYS = { 'event_type' => 'event_type', 'resource_id' => 'resource_id', 'version_no' => 'version_no',
|
||||
'is_latest' => 'is_latest', 'thumbnail' => 'thumbnail_url',
|
||||
'thumbnail_base' => 'thumbnail_base', 'title' => 'title', 'url' => 'url',
|
||||
'tags' => 'tags', 'parents' => 'parents', 'video_ms' => 'video_ms',
|
||||
|
||||
@@ -7,6 +7,12 @@ class CountsExporter
|
||||
attr_reader :format, :resource
|
||||
|
||||
def initialize(params, language_code:)
|
||||
if params.key?('definition')
|
||||
@table_export = CountsTableExport.new(params)
|
||||
@format = @table_export.format
|
||||
@resource = @table_export.resource
|
||||
return
|
||||
end
|
||||
allowed = (CountsQuery::KEYS + CountsStateQuery::KEYS + EXTRA_KEYS).uniq
|
||||
raise ArgumentError if (params.keys - allowed).any?
|
||||
raise ArgumentError unless params.values.all? { _1.is_a?(String) }
|
||||
@@ -42,6 +48,8 @@ class CountsExporter
|
||||
end
|
||||
|
||||
def body
|
||||
return @table_export.body if @table_export
|
||||
|
||||
Enumerator.new do |output|
|
||||
ApplicationRecord.transaction(isolation: :repeatable_read) do
|
||||
first_row = true
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
# Population first, then filters. Previous versions are joined before period filtering.
|
||||
class CountsPopulation
|
||||
attr_reader :definition, :params, :at, :history, :scope, :timezone
|
||||
|
||||
def initialize(params)
|
||||
@params = params
|
||||
@definition = CountsResourceDefinition.new(params.fetch('resource', 'posts'))
|
||||
@history = params.fetch('definition', 'latest') == 'history'
|
||||
raise ArgumentError unless ['latest', 'history'].include?(params.fetch('definition', 'latest'))
|
||||
|
||||
@timezone = TZInfo::Timezone.get(params.fetch('timezone', 'UTC'))
|
||||
@at = timestamp(params['execution_at']) || Time.current.utc
|
||||
raise ArgumentError if at > Time.current + 1.second
|
||||
|
||||
@scope = definition.model.all
|
||||
make_population
|
||||
apply_filters
|
||||
rescue TZInfo::InvalidTimezoneIdentifier
|
||||
raise ArgumentError
|
||||
end
|
||||
|
||||
def timestamp(value, end_date: false)
|
||||
return nil if value.nil? || value == ''
|
||||
raise ArgumentError unless value.is_a?(String)
|
||||
|
||||
if value.match?(/\A\d{4}-\d{2}-\d{2}\z/)
|
||||
date = Date.iso8601(value)
|
||||
date += 1 if end_date
|
||||
timezone.local_to_utc(Time.utc(date.year, date.month, date.day))
|
||||
else
|
||||
raise ArgumentError unless value.match?(/(?:Z|[+-]\d{2}:\d{2})\z/)
|
||||
|
||||
time = Time.iso8601(value).utc
|
||||
raise ArgumentError unless time.year.between?(1900, 2100)
|
||||
|
||||
time
|
||||
end
|
||||
rescue TZInfo::AmbiguousTime, TZInfo::PeriodNotFound, Date::Error
|
||||
raise ArgumentError
|
||||
end
|
||||
|
||||
def latest_sql(row = 'v', cutoff = nil)
|
||||
cutoff ||= history ? at : timestamp(params['at']) || at
|
||||
keys = definition.keys.map { "newer.#{ _1 } = #{ row }.#{ _1 }" }.join(' AND ')
|
||||
bound = definition.model.connection.quote(cutoff)
|
||||
<<~SQL.squish
|
||||
NOT EXISTS (SELECT 1 FROM #{ definition.table } newer
|
||||
WHERE #{ keys } AND newer.version_no > #{ row }.version_no
|
||||
AND newer.#{ definition.time_column } <= #{ bound })
|
||||
SQL
|
||||
end
|
||||
|
||||
def changed_sql(key)
|
||||
column = definition.column(key)
|
||||
"(previous.id IS NOT NULL AND NOT (v.#{ column } <=> previous.#{ column }))"
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def make_population
|
||||
table = definition.table
|
||||
@scope = @scope.from("#{ table } v").select('v.*')
|
||||
keys = definition.keys.map { "previous.#{ _1 } = v.#{ _1 }" }.join(' AND ')
|
||||
@scope = @scope.joins("LEFT JOIN #{ table } previous ON #{ keys } " +
|
||||
'AND previous.version_no = v.version_no - 1')
|
||||
@scope = @scope.where("v.#{ definition.time_column } <= ?", at)
|
||||
unless history
|
||||
point = timestamp(params['at']) || at
|
||||
raise ArgumentError if point > at
|
||||
|
||||
@scope = @scope.where("v.#{ definition.time_column } <= ?", point)
|
||||
.where(Arel.sql(latest_sql('v', point))).where.not('v.event_type' => 'discard')
|
||||
return
|
||||
end
|
||||
first = timestamp(params['from'])
|
||||
last = timestamp(params['through'], end_date: true)
|
||||
raise ArgumentError if first && last && first > last
|
||||
|
||||
@scope = @scope.where('v.created_at >= ?', first) if first
|
||||
if last
|
||||
operator = params['through'].length == 10 ? '<' : '<='
|
||||
@scope = @scope.where("v.created_at #{ operator } ?", last)
|
||||
end
|
||||
end
|
||||
|
||||
def apply_filters
|
||||
['resource_id', 'user'].each do |key|
|
||||
value = params[key == 'user' ? 'user_id' : key]
|
||||
next if value.nil? || value == ''
|
||||
raise ArgumentError unless value.match?(/\A[0-9]+\z/)
|
||||
|
||||
@scope = @scope.where("#{ definition.expression(key) } = ?", value.to_i)
|
||||
end
|
||||
apply_tags if params['tags'].present? || params['tag'].present?
|
||||
conditions = JSON.parse(params.fetch('conditions', '[]'))
|
||||
raise ArgumentError unless conditions.is_a?(Array) && conditions.length <= 100
|
||||
|
||||
conditions.each { apply_condition(_1) }
|
||||
drills = JSON.parse(params.fetch('drill', '[]'))
|
||||
raise ArgumentError unless drills.is_a?(Array) && drills.length <= 100
|
||||
|
||||
drills.each do |entry|
|
||||
key = entry.fetch('field')
|
||||
if key == 'tags'
|
||||
@scope = @scope.where(snapshot_membership(entry.fetch('value')))
|
||||
elsif key == 'changed_field'
|
||||
raise ArgumentError unless history && definition.changes.include?(entry['value'])
|
||||
|
||||
@scope = @scope.where(Arel.sql(changed_sql(entry['value'])))
|
||||
elsif definition.fields[key] == :datetime && entry['grain']
|
||||
@scope = @scope.where(definition.expression(key) => nil) if entry['value'].nil?
|
||||
if entry['value']
|
||||
expression = CountsAxes.time_expression(definition.expression(key),
|
||||
entry['grain'], timezone)
|
||||
@scope = @scope.where("#{ expression } = ?", entry['value'])
|
||||
end
|
||||
else
|
||||
raise ArgumentError unless definition.axes(history).include?(key)
|
||||
|
||||
@scope = @scope.where(definition.expression(key) => entry['value'])
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def apply_condition(entry)
|
||||
raise ArgumentError unless entry.is_a?(Hash)
|
||||
raise ArgumentError if (entry.keys - ['field', 'operator', 'value', 'end', 'blank']).any?
|
||||
|
||||
key = entry.fetch('field')
|
||||
type = definition.fields.fetch(key) { raise ArgumentError }
|
||||
operator = entry.fetch('operator')
|
||||
raise ArgumentError unless CountsResourceDefinition::OPERATORS.fetch(type, []).include?(operator)
|
||||
raise ArgumentError unless [true, false, nil].include?(entry['blank'])
|
||||
|
||||
expression = definition.expression(key)
|
||||
blank = type == :string || type == :enum ?
|
||||
"(#{ expression } IS NULL OR #{ expression } = '')" : "#{ expression } IS NULL"
|
||||
if entry['blank']
|
||||
@scope = @scope.where(Arel.sql(blank))
|
||||
return
|
||||
end
|
||||
values = [entry['value'], entry['end']]
|
||||
raise ArgumentError unless values.all? { _1.nil? || _1.is_a?(String) }
|
||||
return if values.all? { _1.nil? || _1 == '' }
|
||||
|
||||
parsed = values.map do |value|
|
||||
next nil if value.nil? || value == ''
|
||||
|
||||
case type
|
||||
when :integer
|
||||
raise ArgumentError unless value.match?(/\A-?\d+\z/)
|
||||
|
||||
value.to_i
|
||||
when :datetime then timestamp(value)
|
||||
else value
|
||||
end
|
||||
end
|
||||
if operator == 'range'
|
||||
@scope = @scope.where("#{ expression } >= ?", parsed[0]) if parsed[0]
|
||||
@scope = @scope.where("#{ expression } <= ?", parsed[1]) if parsed[1]
|
||||
return
|
||||
end
|
||||
return unless parsed[0]
|
||||
|
||||
sql_operator = { 'eq' => '=', 'ne' => '<>', 'gt' => '>', 'lt' => '<',
|
||||
'ge' => '>=', 'le' => '<=', 'like' => 'LIKE' }.fetch(operator)
|
||||
sql = "#{ expression } #{ sql_operator } ?"
|
||||
sql = "(#{ sql } OR #{ blank })" if operator == 'ne'
|
||||
@scope = @scope.where(sql, parsed[0])
|
||||
end
|
||||
|
||||
def snapshot_membership(value)
|
||||
kind, identifier = value.to_s.split(':', 2)
|
||||
raise ArgumentError unless ['internal', 'external'].include?(kind)
|
||||
raise ArgumentError unless identifier&.match?(/\A[1-9]\d*\z/)
|
||||
|
||||
key = kind == 'internal' ? 'tag_id' : 'external_tag_id'
|
||||
definition.model.sanitize_sql_array([
|
||||
"JSON_CONTAINS(v.tags_json, JSON_OBJECT('#{ key }', CAST(? AS UNSIGNED)))",
|
||||
identifier.to_i])
|
||||
end
|
||||
|
||||
def apply_tags
|
||||
raise ArgumentError unless definition.resource == 'posts'
|
||||
|
||||
if params['tag'].present?
|
||||
@scope = @scope.where(snapshot_membership(
|
||||
"#{ params.fetch('kind', 'internal') }:#{ params['tag'] }"))
|
||||
end
|
||||
return unless params['tags'].present?
|
||||
|
||||
locale = Locale.find_by!(code: params.fetch('locale', 'ja'))
|
||||
refs = PostTagSearch.parse_references(params['tag_refs'], params['tags'].split)
|
||||
raw = params['tags']
|
||||
literals = PostTagSearch.literals(locale, raw, references: refs)
|
||||
predicates = literals.map do |literal|
|
||||
internal, external = PostTagSearch.references(locale, literal)
|
||||
parts = Array(internal).map { snapshot_membership("internal:#{ _1 }") } +
|
||||
Array(external).map { snapshot_membership("external:#{ _1 }") }
|
||||
expression = parts.empty? ? 'FALSE' : "(#{ parts.join(' OR ') })"
|
||||
literal[:negative] ? "NOT #{ expression }" : expression
|
||||
end
|
||||
return if predicates.empty?
|
||||
|
||||
any = params['match'] == 'any' || raw.split.any? { _1.downcase == 'type:or' }
|
||||
@scope = @scope.where(Arel.sql("(#{ predicates.join(any ? ' OR ' : ' AND ') })"))
|
||||
end
|
||||
end
|
||||
@@ -1,7 +1,7 @@
|
||||
# Read-only history queries. SQL identifiers and expressions are server-owned.
|
||||
class CountsQuery
|
||||
AXES = ['month', 'user'].freeze
|
||||
COMMON_COLUMNS = ['resource_id', 'version_no', 'is_latest', 'user', 'created_at'].freeze
|
||||
COMMON_COLUMNS = ['resource_id', 'version_no', 'is_latest', 'event_type', 'user', 'created_at'].freeze
|
||||
COLUMNS = {
|
||||
'posts' => COMMON_COLUMNS + [
|
||||
'thumbnail', 'thumbnail_base', 'title', 'url', 'tags', 'parents', 'video_ms',
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
# Server-owned version identities, attributes and field types. No client SQL is accepted.
|
||||
class CountsResourceDefinition
|
||||
COMMON = { 'resource_id' => :integer, 'version_no' => :integer,
|
||||
'event_type' => :enum, 'user' => :integer, 'created_at' => :datetime }.freeze
|
||||
DEFINITIONS = {
|
||||
'posts' => [PostVersion, ['post_id'], {
|
||||
'title' => :string, 'url' => :string, 'thumbnail_base' => :string,
|
||||
'tags_json' => :json, 'parent_post_ids' => :string, 'video_ms' => :integer,
|
||||
'original_created_from' => :datetime, 'original_created_before' => :datetime }],
|
||||
'tags' => [TagVersion, ['tag_id'], {
|
||||
'name' => :string, 'names' => :json, 'category' => :enum, 'aliases' => :string,
|
||||
'parent_tag_ids' => :string, 'deprecated_at' => :datetime }],
|
||||
'nico_tags' => [NicoTagVersion, ['tag_id'], {
|
||||
'name' => :string, 'linked_tags' => :string }],
|
||||
'tag_descriptions' => [TagDescriptionVersion, ['tag_id', 'language_code'], {
|
||||
'language_code' => :enum, 'body' => :string,
|
||||
'source_language_code' => :enum, 'source_version_no' => :integer }],
|
||||
'tag_groups' => [TagGroupVersion, ['tag_group_id'], {
|
||||
'name' => :string, 'description' => :string,
|
||||
'members' => :json, 'placements' => :json }],
|
||||
'materials' => [MaterialVersion, ['material_id'], {
|
||||
'url' => :string, 'parent_id' => :integer, 'tag_id' => :integer,
|
||||
'tag_name' => :string, 'tag_category' => :enum,
|
||||
'discarded_at' => :datetime, 'updated_at' => :datetime,
|
||||
'updated_by_user_id' => :integer, 'export_paths_json' => :json,
|
||||
'file_blob_id' => :integer, 'file_filename' => :string,
|
||||
'file_content_type' => :string, 'file_byte_size' => :integer,
|
||||
'file_checksum' => :string, 'file_sha256' => :string,
|
||||
'source_kind' => :enum, 'source_uri' => :string, 'source_path' => :string,
|
||||
'source_file_id' => :string, 'normalized_source_key' => :string }] }.freeze
|
||||
OPERATORS = {
|
||||
string: ['eq', 'ne', 'like'], enum: ['eq', 'ne'], boolean: ['eq', 'ne'],
|
||||
integer: ['eq', 'ne', 'gt', 'lt', 'ge', 'le', 'range'],
|
||||
datetime: ['eq', 'ne', 'gt', 'lt', 'ge', 'le', 'range'] }.freeze
|
||||
|
||||
attr_reader :model, :keys, :fields, :resource
|
||||
|
||||
def initialize(resource)
|
||||
@resource = resource
|
||||
@model, @keys, attributes = DEFINITIONS.fetch(resource) { raise ArgumentError }
|
||||
@fields = COMMON.merge(attributes)
|
||||
end
|
||||
|
||||
def table = model.table_name
|
||||
def time_column = 'created_at'
|
||||
def column(key)
|
||||
raise ArgumentError unless fields.key?(key)
|
||||
|
||||
{ 'resource_id' => keys.first, 'user' => 'created_by_user_id' }.fetch(key, key)
|
||||
end
|
||||
|
||||
def expression(key, row = 'v') = "#{ row }.#{ column(key) }"
|
||||
def changes = fields.keys - COMMON.keys - ['updated_at', 'updated_by_user_id']
|
||||
def axes(history)
|
||||
fields.reject { |_key, type| type == :json }.keys +
|
||||
(resource == 'posts' ? ['tags'] : []) + (history ? ['changed_field'] : [])
|
||||
end
|
||||
|
||||
def metadata
|
||||
fields.map do |key, type|
|
||||
{ key:, type:, nullable: model.columns_hash.fetch(column(key)).null,
|
||||
operators: OPERATORS.fetch(type, []) }
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,172 @@
|
||||
require 'csv'
|
||||
|
||||
# Export uses the exact same population and grouped cells as the interactive table.
|
||||
class CountsTableExport
|
||||
EXTRA = ['format', 'columns', 'headers', 'labels', 'children', 'child_sort',
|
||||
'order', 'nested_json', 'summary_columns', 'basis', 'axis', 'latest'].freeze
|
||||
attr_reader :format, :resource
|
||||
|
||||
def initialize(params)
|
||||
raise ArgumentError if (params.keys - CountsTableQuery::KEYS - EXTRA).any?
|
||||
raise ArgumentError unless params.values.all? { _1.is_a?(String) }
|
||||
|
||||
@params = params
|
||||
@labels = JSON.parse(params.fetch('labels', '{}'))
|
||||
raise ArgumentError unless @labels.is_a?(Hash)
|
||||
@selected = params.key?('children') ? params['children'].split(',') : nil
|
||||
@order = params.fetch('order', '').split(',')
|
||||
@child_sort = params.fetch('child_sort', '').split(',').map { _1.split(':', -1) }
|
||||
@child_sort.each do |path, direction, *extra|
|
||||
raise ArgumentError if extra.any? || !['asc', 'desc'].include?(direction)
|
||||
raise ArgumentError unless path&.match?(/\A[a-zA-Z_]+(?:\.[a-zA-Z]+)*\/[a-zA-Z]+\z/)
|
||||
end
|
||||
@format = params.fetch('format', 'json')
|
||||
@resource = params.fetch('resource', 'posts')
|
||||
raise ArgumentError unless ['json', 'csv'].include?(format)
|
||||
|
||||
@query = CountsTableQuery.new(params.slice(*CountsTableQuery::KEYS)
|
||||
.merge('page' => '1', 'limit' => '100'))
|
||||
@summary = @query.mode == 'summary'
|
||||
@columns = params.fetch('columns', '').split(',')
|
||||
@headers = JSON.parse(params.fetch('headers', '[]'))
|
||||
raise ArgumentError unless @headers.is_a?(Array) && @headers.all? { _1.is_a?(String) }
|
||||
|
||||
unless @summary
|
||||
if ['posts', 'tags'].include?(resource)
|
||||
@projection = CountsExportProjection.new(params, resource:, summary: false, tags: false)
|
||||
else
|
||||
allowed = @query.population.definition.fields.keys + ['is_latest']
|
||||
raise ArgumentError if (@columns - allowed).any?
|
||||
raise ArgumentError unless @columns.length == @headers.length && @columns.uniq == @columns
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def body
|
||||
Enumerator.new do |output|
|
||||
ApplicationRecord.transaction(isolation: :repeatable_read) do
|
||||
@language_names = Language.pluck(:code, :name).to_h
|
||||
started = false
|
||||
first = true
|
||||
@query.each_page do |page|
|
||||
unless started
|
||||
metadata = page.except(:versions, :cells, :axis_labels, :column_values).merge(
|
||||
resource:, filters: @params.slice(*CountsTableQuery::KEYS),
|
||||
fields: @query.population.definition.metadata,
|
||||
columns: @columns, column_labels: @headers,
|
||||
order: @params['order'], children: @params['children'],
|
||||
sort: @params.fetch('sort', ''), child_sort: @params['child_sort'])
|
||||
output << (format == 'json' ? "{\"metadata\":#{ metadata.to_json },\"data\":[" : "\uFEFF")
|
||||
if format == 'csv'
|
||||
headers = @summary ? summary_headers : @headers
|
||||
output << CSV.generate_line(headers.map { safe(_1) }, row_sep: "\r\n")
|
||||
end
|
||||
started = true
|
||||
end
|
||||
(page[:cells] || page[:versions]).each do |row|
|
||||
record = project(row)
|
||||
if format == 'json'
|
||||
output << ',' unless first
|
||||
output << record.to_json
|
||||
else
|
||||
values = if @summary
|
||||
row[:values].each_with_index.map do |value, index|
|
||||
page[:axis_labels].fetch("a#{ index }", {}).fetch(value.to_s, value)
|
||||
end + [row[:count]]
|
||||
elsif @projection
|
||||
@projection.csv_record(record, row)
|
||||
else
|
||||
record.map { |key, value| csv_logical(key, value) }
|
||||
end
|
||||
output << CSV.generate_line(values.map { safe(_1) }, row_sep: "\r\n")
|
||||
end
|
||||
first = false
|
||||
end
|
||||
end
|
||||
output << ']}' if format == 'json'
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def summary_headers
|
||||
labels = @labels
|
||||
@query.axes.all.map { labels.fetch(_1['field'], _1['field']) } +
|
||||
[labels.fetch('count', 'count')]
|
||||
end
|
||||
|
||||
def project(row)
|
||||
return { row: row[:values].first(@query.axes.rows.length),
|
||||
column: row[:values].drop(@query.axes.rows.length), count: row[:count] } if @summary
|
||||
return @projection.project(row) if @projection
|
||||
|
||||
@columns.to_h do |column|
|
||||
key = column == 'user' ? 'created_by_user' : column
|
||||
value = row[key]
|
||||
value = project_nested(value, key) if ['members', 'placements', 'export_paths_json'].include?(key)
|
||||
[key, value]
|
||||
end
|
||||
end
|
||||
|
||||
def camel(key)
|
||||
key.gsub(/_([a-z])/) { Regexp.last_match(1).upcase }
|
||||
end
|
||||
|
||||
def project_nested(value, path)
|
||||
rows = path == 'export_paths_json' ? (value || {}).map { |profile, target|
|
||||
{ 'profile' => profile, 'path' => target }
|
||||
} : Array(value)
|
||||
sorts = @child_sort.filter_map do |key, direction|
|
||||
parent, field = key.split('/', 2)
|
||||
[field, direction] if parent == path
|
||||
end
|
||||
rows = rows.each_with_index.sort do |(first, a), (last, b)|
|
||||
compared = 0
|
||||
sorts.each do |field, direction|
|
||||
left = first.find { |key, _| camel(key) == field }&.last
|
||||
right = last.find { |key, _| camel(key) == field }&.last
|
||||
compared = if left.nil? || right.nil?
|
||||
left.nil? == right.nil? ? 0 : (left.nil? ? 1 : -1)
|
||||
else
|
||||
(left.to_s <=> right.to_s) * (direction == 'asc' ? 1 : -1)
|
||||
end
|
||||
break unless compared.zero?
|
||||
end
|
||||
compared.zero? ? a <=> b : compared
|
||||
end.map(&:first)
|
||||
rows.map do |row|
|
||||
fields = row.keys.select { !@selected || @selected.include?("#{ path }.#{ camel(_1) }") }
|
||||
ordered = (@order + fields.map { "#{ path }.#{ camel(_1) }" }).uniq
|
||||
.filter_map { |item| fields.find { "#{ path }.#{ camel(_1) }" == item } }
|
||||
ordered.to_h do |key|
|
||||
item = row[key]
|
||||
item = project_nested(item, "#{ path }.#{ camel(key) }") if item.is_a?(Array)
|
||||
[key, item]
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def csv_logical(key, value)
|
||||
return value ? @labels.fetch('latest', '') : '' if key == 'is_latest'
|
||||
if value.is_a?(Time) || value.is_a?(ActiveSupport::TimeWithZone)
|
||||
return @query.population.timezone.utc_to_local(value.utc).iso8601
|
||||
end
|
||||
if key == 'language_code' || key == 'source_language_code'
|
||||
return @language_names[value]
|
||||
end
|
||||
if value.is_a?(Hash)
|
||||
return value.to_h { |field, item| [@labels.fetch(camel(field.to_s), ''), csv_logical(field, item)] }
|
||||
end
|
||||
return value.map { csv_logical(key, _1) } if value.is_a?(Array)
|
||||
|
||||
@labels.fetch("#{ key }:#{ value }", value)
|
||||
end
|
||||
|
||||
def safe(value)
|
||||
return value if value.is_a?(Numeric)
|
||||
|
||||
text = value.is_a?(Hash) || value.is_a?(Array) ? JSON.generate(value) : value.to_s
|
||||
text.match?(/\A\s*[=+@-]/) ? "'#{ text }" : text
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,226 @@
|
||||
# Common entrypoint for six version resources; legacy endpoints keep their contracts.
|
||||
class CountsTableQuery
|
||||
KEYS = ['resource', 'definition', 'mode', 'at', 'execution_at', 'from', 'through',
|
||||
'timezone', 'rows', 'cols', 'conditions', 'drill', 'tags', 'tag_refs', 'match',
|
||||
'tag', 'kind', 'resource_id', 'user_id', 'page', 'limit', 'sort', 'locale'].freeze
|
||||
attr_reader :population, :axes, :page, :limit, :mode
|
||||
|
||||
def initialize(params)
|
||||
raise ArgumentError if (params.keys - KEYS).any?
|
||||
raise ArgumentError unless params.values.all? { _1.is_a?(String) }
|
||||
|
||||
@params = params
|
||||
@population = CountsPopulation.new(params)
|
||||
@definition = population.definition
|
||||
@connection = @definition.model.connection
|
||||
@axes = CountsAxes.new(population)
|
||||
@page = Integer(params.fetch('page', '1'), 10)
|
||||
@limit = Integer(params.fetch('limit', '20'), 10)
|
||||
@mode = params.fetch('mode', 'summary')
|
||||
raise ArgumentError unless page.between?(1, 1_000_000) && limit.between?(1, 100)
|
||||
raise ArgumentError unless ['summary', 'details'].include?(mode)
|
||||
|
||||
@sort = params.fetch('sort', '').split(',').map { _1.split(':', -1) }
|
||||
allowed = mode == 'summary' ? axes.all.each_index.map { "a#{ _1 }" } + ['count'] :
|
||||
@definition.fields.keys + ['is_latest'] +
|
||||
(@definition.resource == 'posts' ? ['tags', 'parents', 'original_created_at'] : []) +
|
||||
(@definition.resource == 'tags' ? ['parents', 'deprecated'] : [])
|
||||
raise ArgumentError if @sort.map(&:first).uniq.length != @sort.length
|
||||
@sort.each do |key, direction, *extra|
|
||||
measure = mode == 'summary' && key&.match?(/\Ameasure_[0-9a-f]+\z/)
|
||||
measure_values(key) if measure
|
||||
if extra.any? || (!allowed.include?(key) && !measure) || !['asc', 'desc'].include?(direction)
|
||||
raise ArgumentError
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def call
|
||||
ApplicationRecord.transaction do
|
||||
metadata.merge(mode == 'summary' ? summary : details)
|
||||
end
|
||||
end
|
||||
|
||||
def metadata
|
||||
{ page:, limit:, execution_at: population.at.iso8601(6), at: @params['at'].presence,
|
||||
definition: population.history ? 'history' : 'latest',
|
||||
timezone: population.timezone.identifier, rows: axes.rows, cols: axes.columns,
|
||||
fields: @definition.metadata, count: population.scope.except(:select).count }
|
||||
end
|
||||
|
||||
def summary
|
||||
expressions = axes.expressions
|
||||
selected = expressions.each_with_index.map { |expression, index| "#{ expression } AS a#{ index }" }
|
||||
source = population.scope
|
||||
source = source.joins(axes.joins) unless axes.joins.empty?
|
||||
sql = if expressions.empty?
|
||||
source.except(:select).select('COUNT(DISTINCT v.id) AS count').to_sql
|
||||
else
|
||||
source.except(:select).group(*expressions.map { Arel.sql(_1) })
|
||||
.select(Arel.sql((selected + ['COUNT(DISTINCT v.id) AS count']).join(', ')))
|
||||
.to_sql
|
||||
end
|
||||
prefix = "WITH cells AS (#{ sql })"
|
||||
row_keys = axes.rows.each_index.map { "a#{ _1 }" }
|
||||
# Page whole row tuples so a pivot row is never split across display pages.
|
||||
row_selection = row_keys.empty? ? '1 AS singleton' : row_keys.join(', ')
|
||||
row_group = row_keys.empty? ? 'GROUP BY singleton' : "GROUP BY #{ row_keys.join(', ') }"
|
||||
row_sort = @sort.each_with_index.filter_map do |(key, direction), index|
|
||||
if key == 'count' || key.start_with?('measure_')
|
||||
condition = if key == 'count'
|
||||
'TRUE'
|
||||
else
|
||||
measure_values(key).each_with_index.map do |value, offset|
|
||||
"a#{ axes.rows.length + offset } <=> #{ @connection.quote(value) }"
|
||||
end.join(' AND ').presence || 'TRUE'
|
||||
end
|
||||
["MAX(CASE WHEN #{ condition } THEN count END) AS sort#{ index }",
|
||||
"sort#{ index } IS NULL ASC, sort#{ index } #{ direction.upcase }"]
|
||||
elsif row_keys.include?(key)
|
||||
[nil, "#{ key } IS NULL ASC, #{ key } #{ direction.upcase }"]
|
||||
end
|
||||
end
|
||||
extras = row_sort.filter_map(&:first)
|
||||
order_rows = row_sort.map(&:last) + row_keys.map { "#{ _1 } IS NULL, #{ _1 }" }
|
||||
order_rows = ['singleton ASC'] if order_rows.empty?
|
||||
group_sql = "SELECT #{ ([row_selection] + extras).join(', ') } FROM cells #{ row_group }"
|
||||
total = @connection.select_value(
|
||||
"#{ prefix } SELECT COUNT(*) FROM (#{ group_sql }) row_groups").to_i
|
||||
join = row_keys.empty? ? 'TRUE' : row_keys.map { "cells.#{ _1 } <=> page_rows.#{ _1 }" }.join(' AND ')
|
||||
order_cols = @sort.filter_map do |key, direction|
|
||||
"cells.#{ key } #{ direction.upcase }" if key.match?(/\Aa[0-9]+\z/) && !row_keys.include?(key)
|
||||
end
|
||||
records = @connection.select_all(<<~SQL)
|
||||
#{ prefix }, row_groups AS (#{ group_sql }),
|
||||
numbered AS (SELECT row_groups.*,
|
||||
ROW_NUMBER() OVER (ORDER BY #{ order_rows.join(', ') }) AS position FROM row_groups),
|
||||
page_rows AS (SELECT * FROM numbered
|
||||
WHERE position > #{ (page - 1) * limit } AND position <= #{ page * limit })
|
||||
SELECT cells.* FROM cells JOIN page_rows ON #{ join }
|
||||
ORDER BY #{ (['page_rows.position'] + order_cols + axes.all.each_index.map { "cells.a#{ _1 }" }).join(', ') }
|
||||
SQL
|
||||
column_keys = axes.columns.each_index.map { "a#{ axes.rows.length + _1 }" }
|
||||
column_values = if column_keys.empty?
|
||||
[[]]
|
||||
else
|
||||
@connection.select_rows(
|
||||
"#{ prefix } SELECT DISTINCT #{ column_keys.join(', ') } FROM cells " +
|
||||
"ORDER BY #{ column_keys.map { "#{ _1 } IS NULL, #{ _1 }" }.join(', ') }")
|
||||
end
|
||||
{ group_count: total, column_values:, cells: records.map do |row|
|
||||
{ values: axes.all.each_index.map { row["a#{ _1 }"] }, count: row['count'].to_i }
|
||||
end, axis_labels: axis_labels(records, column_values) }
|
||||
end
|
||||
|
||||
def details
|
||||
query = population.scope.select(Arel.sql(
|
||||
"(#{ population.latest_sql }) AS counts_latest"))
|
||||
order = sort_sql { detail_expression(_1) }
|
||||
query = query.order(Arel.sql((order + ['v.id DESC']).join(', ')))
|
||||
.limit(limit).offset((page - 1) * limit)
|
||||
rows = query.to_a
|
||||
if ['posts', 'tags'].include?(@definition.resource)
|
||||
values = CountsVersionRepr.many(rows, resource: @definition.resource,
|
||||
language_code: locale.language_code,
|
||||
at: population.history ? population.at : population.timestamp(@params['at']) || population.at)
|
||||
values.each_with_index { |value, index| value[:event_type] = rows[index].event_type }
|
||||
else
|
||||
users = User.where(id: rows.map(&:created_by_user_id)).pluck(:id, :name).to_h
|
||||
values = rows.map do |row|
|
||||
attributes = @definition.fields.keys.to_h do |key|
|
||||
[key, row.public_send(@definition.column(key))]
|
||||
end
|
||||
attributes.merge('id' => row.id, 'resource_id' => row[@definition.keys.first],
|
||||
'is_latest' => row.attributes['counts_latest'].to_i == 1,
|
||||
'created_by_user' => row.created_by_user_id &&
|
||||
{ id: row.created_by_user_id, name: users[row.created_by_user_id] })
|
||||
end
|
||||
end
|
||||
{ versions: values }
|
||||
end
|
||||
|
||||
def each_page
|
||||
total = nil
|
||||
loop do
|
||||
result = call
|
||||
total ||= mode == 'summary' ? result[:group_count] : result[:count]
|
||||
yield result
|
||||
break if page * limit >= total
|
||||
|
||||
@page += 1
|
||||
end
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def locale = @locale ||= Locale.find_by!(code: @params.fetch('locale', 'ja'))
|
||||
|
||||
def measure_values(key)
|
||||
value = JSON.parse([key.delete_prefix('measure_')].pack('H*'))
|
||||
unless (value.is_a?(Array) && value.length == axes.columns.length &&
|
||||
value.all? { _1.nil? || _1.is_a?(String) || _1.is_a?(Numeric) })
|
||||
raise ArgumentError
|
||||
end
|
||||
value
|
||||
end
|
||||
|
||||
def detail_expression(key)
|
||||
return "(#{ population.latest_sql })" if key == 'is_latest'
|
||||
aliases = { 'tags' => 'tags_json', 'parents' => @definition.resource == 'posts' ?
|
||||
'parent_post_ids' : 'parent_tag_ids', 'deprecated' => 'deprecated_at',
|
||||
'original_created_at' => 'original_created_from' }
|
||||
field = aliases.fetch(key, key)
|
||||
expression = @definition.expression(field)
|
||||
@definition.fields[field] == :json ? "CAST(#{ expression } AS CHAR) COLLATE utf8mb4_bin" : expression
|
||||
end
|
||||
|
||||
def sort_sql
|
||||
@sort.flat_map do |key, direction|
|
||||
expression = yield key
|
||||
["(#{ expression }) IS NULL ASC", "#{ expression } #{ direction.upcase }"]
|
||||
end
|
||||
end
|
||||
|
||||
def axis_labels(records, column_values)
|
||||
result = {}
|
||||
axes.all.each_with_index do |axis, index|
|
||||
values = records.map { _1["a#{ index }"] }
|
||||
if index >= axes.rows.length
|
||||
values += column_values.map { _1[index - axes.rows.length] }
|
||||
end
|
||||
values = values.compact.uniq
|
||||
labels = case axis['field']
|
||||
when 'user'
|
||||
User.where(id: values).pluck(:id, :name).to_h.transform_keys(&:to_s)
|
||||
when 'language_code', 'source_language_code'
|
||||
Language.where(code: values).pluck(:code, :name).to_h
|
||||
when 'tags'
|
||||
tag_labels(values)
|
||||
else {}
|
||||
end
|
||||
result["a#{ index }"] = labels
|
||||
end
|
||||
result
|
||||
end
|
||||
|
||||
def tag_labels(values)
|
||||
internal = values.grep(/\Ainternal:/).map { _1.split(':').last.to_i }
|
||||
external = values.grep(/\Aexternal:/).map { _1.split(':').last.to_i }
|
||||
point = population.timestamp(@params['at']) || population.at
|
||||
labels = {}
|
||||
[[TagVersion, internal, 'internal'], [NicoTagVersion, external, 'external']].each do |model, ids, kind|
|
||||
next if ids.empty?
|
||||
|
||||
table = model.table_name
|
||||
model.where(tag_id: ids).where('created_at <= ?', point)
|
||||
.where("NOT EXISTS (SELECT 1 FROM #{ table } newer " +
|
||||
"WHERE newer.tag_id = #{ table }.tag_id AND newer.created_at <= ? " +
|
||||
"AND newer.version_no > #{ table }.version_no)", point).each do |row|
|
||||
names = row.respond_to?(:names) ? row.names : nil
|
||||
name = names&.dig(locale.language_code, 'name') || row.name
|
||||
labels["#{ kind }:#{ row.tag_id }"] = name
|
||||
end
|
||||
end
|
||||
labels
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,42 @@
|
||||
# Shared literal parsing and reference resolution for current and snapshot post search.
|
||||
class PostTagSearch
|
||||
def self.parse_references(raw, literals)
|
||||
return nil if raw.blank?
|
||||
|
||||
references = JSON.parse(raw)
|
||||
valid = references.is_a?(Array) && references.all? do |reference|
|
||||
next false unless reference.is_a?(Hash) && reference['name'].is_a?(String)
|
||||
|
||||
ids = [reference['tag_id'], reference['external_tag_id']].compact
|
||||
((ids.any? || reference['missing'] == true) &&
|
||||
ids.all? { _1.is_a?(Integer) && _1.positive? } &&
|
||||
(!reference.key?('missing') || (reference['missing'] == true && ids.empty?)) &&
|
||||
(!reference.key?('negative') || [true, false].include?(reference['negative'])))
|
||||
end
|
||||
raise ArgumentError unless valid
|
||||
|
||||
references.map { _1['name'] } == literals ? references : nil
|
||||
end
|
||||
|
||||
def self.literals(locale, raw, references: nil)
|
||||
raw.split.each_with_index.filter_map do |token, index|
|
||||
next if token.downcase == 'type:or'
|
||||
|
||||
reference = references&.[](index)
|
||||
name = token.sub(/\Anot:/i, '')
|
||||
name = TagName.canonicalise(locale, name).first unless reference
|
||||
negative = reference&.key?('negative') ? reference['negative'] :
|
||||
token.downcase.start_with?('not:')
|
||||
{ name:, negative:, reference: }
|
||||
end
|
||||
end
|
||||
|
||||
def self.references(locale, literal)
|
||||
reference = literal[:reference]
|
||||
return [reference['tag_id'], reference['external_tag_id']] if reference
|
||||
|
||||
[TagName.where(language_code: locale.language_code, name: literal[:name])
|
||||
.distinct.pluck(:tag_id),
|
||||
ExternalTag.where("CONCAT(platform, ':', name) = ?", literal[:name]).pluck(:id)]
|
||||
end
|
||||
end
|
||||
新しいイシューから参照
ユーザーをブロックする